implement block appender for fast sync (#1164)

core/consensus/babe/block_appender.hpp

@@ -0,0 +1,23 @@
+/**
+ * Copyright Soramitsu Co., Ltd. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#ifndef KAGOME_CONSENSUS_BLOCKAPPENDER
+#define KAGOME_CONSENSUS_BLOCKAPPENDER
+
+#include "outcome/outcome.hpp"
+#include "primitives/block_data.hpp"
+
+namespace kagome::consensus {
+
+  class BlockAppender {
+   public:
+    virtual ~BlockAppender() = default;
+
+    virtual outcome::result<void> appendBlock(primitives::BlockData &&b) = 0;
+  };
+
+}  // namespace kagome::consensus
+
+#endif  // KAGOME_CONSENSUS_BLOCKAPPENDER

core/consensus/babe/impl/CMakeLists.txt

@@ -30,6 +30,19 @@ target_link_libraries(babe_digests_util
     block_storage_error
     )
 
+add_library(block_appender
+    block_appender_impl.cpp
+    )
+target_link_libraries(block_appender
+    logger
+    primitives
+    scale::scale
+    block_tree_error
+    threshold_util
+    babe_digests_util
+    blockchain_common
+    )
+
 add_library(block_executor
     block_executor_impl.cpp
     )

core/consensus/babe/impl/block_appender_impl.cpp

@@ -0,0 +1,220 @@
+/**
+ * Copyright Soramitsu Co., Ltd. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include "consensus/babe/impl/block_appender_impl.hpp"
+
+#include <chrono>
+
+#include "blockchain/block_tree_error.hpp"
+#include "blockchain/impl/common.hpp"
+#include "consensus/babe/impl/babe_digests_util.hpp"
+#include "consensus/babe/impl/threshold_util.hpp"
+#include "consensus/babe/types/slot.hpp"
+#include "network/helpers/peer_id_formatter.hpp"
+#include "primitives/common.hpp"
+#include "scale/scale.hpp"
+#include "transaction_pool/transaction_pool_error.hpp"
+
+OUTCOME_CPP_DEFINE_CATEGORY(kagome::consensus, BlockAppenderImpl::Error, e) {
+  using E = kagome::consensus::BlockAppenderImpl::Error;
+  switch (e) {
+    case E::INVALID_BLOCK:
+      return "Invalid block";
+    case E::PARENT_NOT_FOUND:
+      return "Parent not found";
+  }
+  return "Unknown error";
+}
+
+namespace {
+  constexpr const char *kBlockExecutionTime =
+      "kagome_block_verification_and_import_time";
+}
+
+namespace kagome::consensus {
+
+  BlockAppenderImpl::BlockAppenderImpl(
+      std::shared_ptr<blockchain::BlockTree> block_tree,
+      std::shared_ptr<primitives::BabeConfiguration> configuration,
+      std::shared_ptr<BlockValidator> block_validator,
+      std::shared_ptr<grandpa::Environment> grandpa_environment,
+      std::shared_ptr<crypto::Hasher> hasher,
+      std::shared_ptr<authority::AuthorityUpdateObserver>
+          authority_update_observer,
+      std::shared_ptr<BabeUtil> babe_util)
+      : block_tree_{std::move(block_tree)},
+        babe_configuration_{std::move(configuration)},
+        block_validator_{std::move(block_validator)},
+        grandpa_environment_{std::move(grandpa_environment)},
+        hasher_{std::move(hasher)},
+        authority_update_observer_{std::move(authority_update_observer)},
+        babe_util_(std::move(babe_util)),
+        logger_{log::createLogger("BlockAppender", "block_appender")} {
+    BOOST_ASSERT(block_tree_ != nullptr);
+    BOOST_ASSERT(babe_configuration_ != nullptr);
+    BOOST_ASSERT(block_validator_ != nullptr);
+    BOOST_ASSERT(grandpa_environment_ != nullptr);
+    BOOST_ASSERT(hasher_ != nullptr);
+    BOOST_ASSERT(authority_update_observer_ != nullptr);
+    BOOST_ASSERT(babe_util_ != nullptr);
+    BOOST_ASSERT(logger_ != nullptr);
+  }
+
+  outcome::result<void> BlockAppenderImpl::appendBlock(
+      primitives::BlockData &&b) {
+    if (not b.header.has_value()) {
+      logger_->warn("Skipping a block without header");
+      return Error::INVALID_BLOCK;
+    }
+    auto &header = b.header.value();
+
+    if (auto header_res = block_tree_->getBlockHeader(header.parent_hash);
+        header_res.has_error()
+        && header_res.error() == blockchain::BlockTreeError::HEADER_NOT_FOUND) {
+      logger_->warn("Skipping a block with unknown parent");
+      return Error::PARENT_NOT_FOUND;
+    } else if (header_res.has_error()) {
+      return header_res.as_failure();
+    }
+
+    // get current time to measure performance if block execution
+    auto t_start = std::chrono::high_resolution_clock::now();
+
+    auto block_hash = hasher_->blake2b_256(scale::encode(header).value());
+
+    bool block_already_exists = false;
+
+    // check if block header already exists. If so, do not append
+    if (auto header_res = block_tree_->getBlockHeader(block_hash);
+        header_res.has_value()) {
+      SL_DEBUG(logger_,
+               "Skip existing block: {}",
+               primitives::BlockInfo(header.number, block_hash));
+
+      OUTCOME_TRY(block_tree_->addExistingBlock(block_hash, header));
+      block_already_exists = true;
+    } else if (header_res.error() != blockchain::BlockTreeError::HEADER_NOT_FOUND) {
+      return header_res.as_failure();
+    }
+
+    primitives::Block block{.header = std::move(header)};
+
+    OUTCOME_TRY(babe_digests, getBabeDigests(block.header));
+
+    const auto &babe_header = babe_digests.second;
+
+    auto slot_number = babe_header.slot_number;
+    auto epoch_number = babe_util_->slotToEpoch(slot_number);
+
+    logger_->info(
+        "Applying block {} ({} in slot {}, epoch {})",  //
+        primitives::BlockInfo(block.header.number, block_hash),
+        babe_header.slotType() == SlotType::Primary          ? "primary"
+        : babe_header.slotType() == SlotType::SecondaryVRF   ? "secondary-vrf"
+        : babe_header.slotType() == SlotType::SecondaryPlain ? "secondary-plain"
+                                                             : "unknown",
+        slot_number,
+        epoch_number);
+
+    OUTCOME_TRY(
+        this_block_epoch_descriptor,
+        block_tree_->getEpochDigest(epoch_number, block.header.parent_hash));
+
+    SL_TRACE(logger_,
+             "Actual epoch digest to apply block {} (slot {}, epoch {}). "
+             "Randomness: {}",
+             primitives::BlockInfo(block.header.number, block_hash),
+             slot_number,
+             epoch_number,
+             this_block_epoch_descriptor.randomness);
+
+    auto threshold = calculateThreshold(babe_configuration_->leadership_rate,
+                                        this_block_epoch_descriptor.authorities,
+                                        babe_header.authority_index);
+
+    OUTCOME_TRY(block_validator_->validateHeader(
+        block.header,
+        epoch_number,
+        this_block_epoch_descriptor.authorities[babe_header.authority_index].id,
+        threshold,
+        this_block_epoch_descriptor.randomness));
+
+    if (auto next_epoch_digest_res = getNextEpochDigest(block.header)) {
+      auto &next_epoch_digest = next_epoch_digest_res.value();
+      SL_VERBOSE(logger_,
+                 "Next epoch digest has gotten in block {} (slot {}). "
+                 "Randomness: {}",
+                 slot_number,
+                 primitives::BlockInfo(block.header.number, block_hash),
+                 next_epoch_digest.randomness);
+    }
+
+    auto block_without_seal_digest = block;
+
+    // block should be applied without last digest which contains the seal
+    block_without_seal_digest.header.digest.pop_back();
+
+    auto parent = block_tree_->getBlockHeader(block.header.parent_hash).value();
+
+    if (not block_already_exists) {
+      OUTCOME_TRY(block_tree_->addBlockHeader(block.header));
+    }
+
+    // observe possible changes of authorities
+    // (must be done strictly after block will be added)
+    for (auto &digest_item : block_without_seal_digest.header.digest) {
+      auto res = visit_in_place(
+          digest_item,
+          [&](const primitives::Consensus &consensus_message)
+              -> outcome::result<void> {
+            return authority_update_observer_->onConsensus(
+                primitives::BlockInfo{block.header.number, block_hash},
+                consensus_message);
+          },
+          [](const auto &) { return outcome::success(); });
+      if (res.has_error()) {
+        rollbackBlock(block_hash);
+        return res.as_failure();
+      }
+    }
+
+    // apply justification if any (must be done strictly after block will be
+    // added and his consensus-digests will be handled)
+    if (b.justification.has_value()) {
+      SL_VERBOSE(logger_,
+                 "Justification received for block {}",
+                 primitives::BlockInfo(block.header.number, block_hash));
+      auto res = grandpa_environment_->applyJustification(
+          primitives::BlockInfo(block.header.number, block_hash),
+          b.justification.value());
+      if (res.has_error()) {
+        rollbackBlock(block_hash);
+        return res.as_failure();
+      }
+    }
+
+    auto t_end = std::chrono::high_resolution_clock::now();
+
+    logger_->info(
+        "Imported block {} within {} ms",
+        primitives::BlockInfo(block.header.number, block_hash),
+        std::chrono::duration_cast<std::chrono::milliseconds>(t_end - t_start)
+            .count());
+
+    return outcome::success();
+  }
+
+  void BlockAppenderImpl::rollbackBlock(
+      const primitives::BlockHash &block_hash) {
+    auto removal_res = block_tree_->removeLeaf(block_hash);
+    if (removal_res.has_error()) {
+      SL_WARN(logger_,
+              "Rolling back of block {} is failed: {}",
+              block_hash,
+              removal_res.error().message());
+    }
+  }
+
+  }  // namespace kagome::consensus

core/consensus/babe/impl/block_appender_impl.hpp

@@ -0,0 +1,63 @@
+/**
+ * Copyright Soramitsu Co., Ltd. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#ifndef KAGOME_CONSENSUS_BLOCKAPPENDERIMPL
+#define KAGOME_CONSENSUS_BLOCKAPPENDERIMPL
+
+#include "consensus/babe/block_appender.hpp"
+
+#include <libp2p/peer/peer_id.hpp>
+
+#include "blockchain/block_tree.hpp"
+#include "clock/timer.hpp"
+#include "consensus/authority/authority_update_observer.hpp"
+#include "consensus/babe/babe_util.hpp"
+#include "consensus/grandpa/environment.hpp"
+#include "consensus/validation/block_validator.hpp"
+#include "crypto/hasher.hpp"
+#include "log/logger.hpp"
+#include "primitives/babe_configuration.hpp"
+#include "primitives/block_header.hpp"
+
+namespace kagome::consensus {
+
+  class BlockAppenderImpl
+      : public BlockAppender,
+        public std::enable_shared_from_this<BlockAppenderImpl> {
+   public:
+    enum class Error { INVALID_BLOCK = 1, PARENT_NOT_FOUND };
+
+    BlockAppenderImpl(
+        std::shared_ptr<blockchain::BlockTree> block_tree,
+        std::shared_ptr<primitives::BabeConfiguration> configuration,
+        std::shared_ptr<BlockValidator> block_validator,
+        std::shared_ptr<grandpa::Environment> grandpa_environment,
+        std::shared_ptr<crypto::Hasher> hasher,
+        std::shared_ptr<authority::AuthorityUpdateObserver>
+            authority_update_observer,
+        std::shared_ptr<BabeUtil> babe_util);
+
+    outcome::result<void> appendBlock(primitives::BlockData &&b) override;
+
+   private:
+    void rollbackBlock(const primitives::BlockHash &block_hash);
+
+    std::shared_ptr<blockchain::BlockTree> block_tree_;
+    std::shared_ptr<primitives::BabeConfiguration> babe_configuration_;
+    std::shared_ptr<BlockValidator> block_validator_;
+    std::shared_ptr<grandpa::Environment> grandpa_environment_;
+    std::shared_ptr<crypto::Hasher> hasher_;
+    std::shared_ptr<authority::AuthorityUpdateObserver>
+        authority_update_observer_;
+    std::shared_ptr<BabeUtil> babe_util_;
+
+    log::Logger logger_;
+  };
+
+}  // namespace kagome::consensus
+
+OUTCOME_HPP_DECLARE_ERROR(kagome::consensus, BlockAppenderImpl::Error);
+
+#endif  // KAGOME_CONSENSUS_BLOCKAPPENDERIMPL

core/injector/CMakeLists.txt

@@ -32,6 +32,7 @@ target_link_libraries(application_injector
     binaryen_module_factory
     bip39_provider
     block_announce_protocol
+    block_appender
     block_builder_api
     block_executor
     block_header_repository

core/injector/application_injector.cpp

@@ -60,6 +60,7 @@
 #include "consensus/babe/impl/babe_impl.hpp"
 #include "consensus/babe/impl/babe_lottery_impl.hpp"
 #include "consensus/babe/impl/babe_util_impl.hpp"
+#include "consensus/babe/impl/block_appender_impl.hpp"
 #include "consensus/babe/impl/block_executor_impl.hpp"
 #include "consensus/babe/impl/consistency_keeper_impl.hpp"
 #include "consensus/grandpa/impl/environment_impl.hpp"
@@ -1265,6 +1266,7 @@ namespace {
         di::bind<network::PeerManager>.to(
             [](auto const &injector) { return get_peer_manager(injector); }),
         di::bind<network::Router>.template to<network::RouterLibp2p>(),
+        di::bind<consensus::BlockAppender>.template to<consensus::BlockAppenderImpl>(),
         di::bind<consensus::BlockExecutor>.to(
             [](auto const &injector) { return get_block_executor(injector); }),
         di::bind<consensus::grandpa::Grandpa>.to(

core/log/configurator.cpp

@@ -50,6 +50,7 @@ namespace kagome::log {
               - name: babe
                 children:
                   - name: babe_lottery
+                  - name: block_appender
                   - name: block_executor
                   - name: block_validator
               - name: grandpa