synchronizer_impl.hpp

Go to the documentation of this file.

#ifndef KAGOME_NETWORK_SYNCHRONIZERIMPL
#define KAGOME_NETWORK_SYNCHRONIZERIMPL

#include "network/synchronizer.hpp"

#include <atomic>
#include <mutex>
#include <queue>

#include <libp2p/basic/scheduler.hpp>

#include "application/app_state_manager.hpp"
#include "consensus/babe/block_appender.hpp"
#include "consensus/babe/block_executor.hpp"
#include "metrics/metrics.hpp"
#include "network/router.hpp"
#include "storage/buffer_map_types.hpp"
#include "telemetry/service.hpp"

namespace kagome::application {
  class AppConfiguration;
}

namespace kagome::storage::changes_trie {
  class ChangesTracker;
}

namespace kagome::storage::trie {
  class PersistentTrieBatch;
  class TrieSerializer;
  class TrieStorage;
}  // namespace kagome::storage::trie

namespace kagome::network {

  class SynchronizerImpl
      : public Synchronizer,
        public std::enable_shared_from_this<SynchronizerImpl> {
   public:
    static constexpr size_t kMinPreloadedBlockAmount = 256;

    static constexpr size_t kMinPreloadedBlockAmountForFastSyncing =
        kMinPreloadedBlockAmount * 40;

    static constexpr size_t kMaxDistanceToBlockForSubscription =
        kMinPreloadedBlockAmount * 2;

    static constexpr std::chrono::milliseconds kRecentnessDuration =
        std::chrono::seconds(60);

    enum class Error {
      SHUTTING_DOWN = 1,
      EMPTY_RESPONSE,
      RESPONSE_WITHOUT_BLOCK_HEADER,
      RESPONSE_WITHOUT_BLOCK_BODY,
      DISCARDED_BLOCK,
      WRONG_ORDER,
      INVALID_HASH,
      ALREADY_IN_QUEUE,
      PEER_BUSY,
      ARRIVED_TOO_EARLY,
      DUPLICATE_REQUEST,
    };

    SynchronizerImpl(
        const application::AppConfiguration &app_config,
        std::shared_ptr<application::AppStateManager> app_state_manager,
        std::shared_ptr<blockchain::BlockTree> block_tree,
        std::shared_ptr<storage::changes_trie::ChangesTracker> changes_tracker,
        std::shared_ptr<consensus::BlockAppender> block_appender,
        std::shared_ptr<consensus::BlockExecutor> block_executor,
        std::shared_ptr<storage::trie::TrieSerializer> serializer,
        std::shared_ptr<storage::trie::TrieStorage> storage,
        std::shared_ptr<network::Router> router,
        std::shared_ptr<libp2p::basic::Scheduler> scheduler,
        std::shared_ptr<crypto::Hasher> hasher,
        std::shared_ptr<storage::BufferStorage> buffer_storage);

    bool prepare();

    bool start();

    void stop();

    bool syncByBlockInfo(const primitives::BlockInfo &block_info,
                         const libp2p::peer::PeerId &peer_id,
                         SyncResultHandler &&handler,
                         bool subscribe_to_block) override;

    bool syncByBlockHeader(const primitives::BlockHeader &header,
                           const libp2p::peer::PeerId &peer_id,
                           SyncResultHandler &&handler) override;

    // See loadJustifications
    void syncMissingJustifications(const PeerId &peer_id,
                                   primitives::BlockInfo target_block,
                                   std::optional<uint32_t> limit,
                                   SyncResultHandler &&handler) override;

    void syncState(const libp2p::peer::PeerId &peer_id,
                   const primitives::BlockInfo &block,
                   SyncResultHandler &&handler) override;

    void findCommonBlock(const libp2p::peer::PeerId &peer_id,
                         primitives::BlockNumber lower,
                         primitives::BlockNumber upper,
                         primitives::BlockNumber hint,
                         SyncResultHandler &&handler,
                         std::map<primitives::BlockNumber,
                                  primitives::BlockHash> &&observed = {});

    void loadBlocks(const libp2p::peer::PeerId &peer_id,
                    primitives::BlockInfo from,
                    SyncResultHandler &&handler);

    void loadJustifications(const libp2p::peer::PeerId &peer_id,
                            primitives::BlockInfo target_block,
                            std::optional<uint32_t> limit,
                            SyncResultHandler &&handler);

    bool hasIncompleteRequestOfStateSync() const override {
      return state_sync_request_.has_value();
    }

   private:
    bool subscribeToBlock(const primitives::BlockInfo &block_info,
                          SyncResultHandler &&handler);

    void notifySubscribers(const primitives::BlockInfo &block_info,
                           outcome::result<void> res);

    void askNextPortionOfBlocks();

    void applyNextBlock();

    void applyNextJustification();

    size_t discardBlock(const primitives::BlockHash &block);

    void prune(const primitives::BlockInfo &finalized_block);

    void scheduleRecentRequestRemoval(
        const libp2p::peer::PeerId &peer_id,
        const BlocksRequest::Fingerprint &fingerprint);

    std::shared_ptr<application::AppStateManager> app_state_manager_;
    std::shared_ptr<blockchain::BlockTree> block_tree_;
    std::shared_ptr<storage::changes_trie::ChangesTracker>
        trie_changes_tracker_;
    std::shared_ptr<consensus::BlockAppender> block_appender_;
    std::shared_ptr<consensus::BlockExecutor> block_executor_;
    std::shared_ptr<storage::trie::TrieSerializer> serializer_;
    std::shared_ptr<storage::trie::TrieStorage> storage_;
    std::shared_ptr<network::Router> router_;
    std::shared_ptr<libp2p::basic::Scheduler> scheduler_;
    std::shared_ptr<crypto::Hasher> hasher_;
    std::shared_ptr<storage::BufferStorage> buffer_storage_;

    application::AppConfiguration::SyncMethod sync_method_;

    // Metrics
    metrics::RegistryPtr metrics_registry_ = metrics::createRegistry();
    metrics::Gauge *metric_import_queue_length_;

    log::Logger log_ = log::createLogger("Synchronizer", "synchronizer");
    telemetry::Telemetry telemetry_ = telemetry::createTelemetryService();

    std::atomic_bool state_sync_request_in_progress_ = false;
    std::optional<network::StateRequest> state_sync_request_;
    std::optional<primitives::BlockInfo> state_sync_on_block_;

    bool node_is_shutting_down_ = false;

    struct KnownBlock {
      primitives::BlockData data;
      std::set<libp2p::peer::PeerId> peers;
    };

    // Already known (enqueued) but is not applied yet
    std::unordered_map<primitives::BlockHash, KnownBlock> known_blocks_;

    std::optional<primitives::BlockInfo> sync_block_;

    // Blocks grouped by number
    std::multimap<primitives::BlockNumber, primitives::BlockHash> generations_;

    // Links parent->child
    std::unordered_multimap<primitives::BlockHash, primitives::BlockHash>
        ancestry_;

    // Loaded justifications to apply
    using JustificationPair =
        std::pair<primitives::BlockInfo, primitives::Justification>;
    std::queue<JustificationPair> justifications_;
    std::mutex justifications_mutex_;

    // BlockNumber of blocks (aka height) that is potentially best now
    primitives::BlockNumber watched_blocks_number_{};

    // Handlers what will be called when block is apply
    std::unordered_multimap<primitives::BlockHash, SyncResultHandler>
        watched_blocks_;

    std::multimap<primitives::BlockInfo, SyncResultHandler> subscriptions_;

    std::atomic_bool applying_in_progress_ = false;
    std::atomic_bool asking_blocks_portion_in_progress_ = false;
    std::set<libp2p::peer::PeerId> busy_peers_;

    std::set<std::tuple<libp2p::peer::PeerId, BlocksRequest::Fingerprint>>
        recent_requests_;

    std::unordered_map<
        storage::trie::RootHash,
        std::tuple<common::Buffer,
                   unsigned,
                   std::shared_ptr<storage::trie::PersistentTrieBatch>>>
        batches_store_;
    size_t entries_{0};
  };

}  // namespace kagome::network

OUTCOME_HPP_DECLARE_ERROR(kagome::network, SynchronizerImpl::Error)

#endif  //  KAGOME_NETWORK_SYNCHRONIZERIMPL