Dynamic trie pruning (#1556)

* Prune unused trie nodes on old and discarded block states

CMakeLists.txt

@@ -95,6 +95,7 @@ if("${CMAKE_CXX_COMPILER_ID}" MATCHES "^(AppleClang|Clang|GNU)$")
     add_flag(-Werror=sign-compare)     # warn the user if they compare a signed and unsigned numbers
     add_flag(-Werror=reorder)          # field '$1' will be initialized after field '$2'
     add_flag(-Werror=mismatched-tags)  # warning: class '$1' was previously declared as struct
+    add_flag(-Werror=switch)  # unhandled values in a switch statement
     # cmake-format: on
   else()
     # promote to errors
@@ -104,6 +105,7 @@ if("${CMAKE_CXX_COMPILER_ID}" MATCHES "^(AppleClang|Clang|GNU)$")
     add_flag(-Werror-non-virtual-dtor) # warn the user if a class with virtual functions has a non-virtual destructor. This helps catch hard to track down memory errors
     add_flag(-Werror-sign-compare)     # warn the user if they compare a signed and unsigned numbers
     add_flag(-Werror-reorder)          # field '$1' will be initialized after field '$2'
+    add_flag(-Werror-switch)          # unhandled values in a switch statement
     # cmake-format: on
   endif()
 elseif("${CMAKE_CXX_COMPILER_ID}" STREQUAL "MSVC")
@@ -112,6 +114,9 @@ elseif("${CMAKE_CXX_COMPILER_ID}" STREQUAL "MSVC")
   # https://github.com/lefticus/cppbestpractices/blob/master/02-Use_the_Tools_Available.md#msvc
 endif()
 
+print("C flags: ${CMAKE_C_FLAGS}")
+print("CXX flags: ${CMAKE_CXX_FLAGS}")
+
 if(COVERAGE)
   include(cmake/coverage.cmake)
 endif()
@@ -164,6 +169,7 @@ kagome_install_setup(
   core/scale
   core/storage
   core/subscription
+  core/telemetry
   core/utils)
 
 include(CMakePackageConfigHelpers)

core/api/service/child_state/impl/child_state_api_impl.cpp

@@ -129,7 +129,8 @@ namespace kagome::api {
                 storage_->getEphemeralBatchAt(child_root_hash));
     auto res = child_storage_trie_reader->tryGet(key);
     return common::map_result_optional(
-        std::move(res), [](common::BufferOrView &&r) { return r.into(); });
+        std::move(res),
+        [](common::BufferOrView &&r) { return r.intoBuffer(); });
   }
 
   outcome::result<std::optional<primitives::BlockHash>>

core/api/service/state/impl/state_api_impl.cpp

@@ -122,7 +122,7 @@ namespace kagome::api {
     OUTCOME_TRY(trie_reader, storage_->getEphemeralBatchAt(header.state_root));
     auto res = trie_reader->tryGet(key);
     return common::map_result_optional(
-        std::move(res), [](common::BufferOrView &&r) { return r.into(); });
+        std::move(res), [](common::BufferOrView &&r) { return r.intoBuffer(); });
   }
 
   outcome::result<std::vector<StateApiImpl::StorageChangeSet>>
@@ -164,7 +164,7 @@ namespace kagome::api {
         OUTCOME_TRY(opt_get, batch->tryGet(key));
         auto opt_value = common::map_optional(
             std::move(opt_get),
-            [](common::BufferOrView &&r) { return r.into(); });
+            [](common::BufferOrView &&r) { return r.intoBuffer(); });
         auto it = last_values.find(key);
         if (it == last_values.end() || it->second != opt_value) {
           change.changes.push_back(StorageChangeSet::Change{key, opt_value});

core/application/CMakeLists.txt

@@ -37,6 +37,7 @@ target_link_libraries(app_config
     chain_spec
     build_version
     )
+kagome_install(app_config)
 
 add_library(chain_spec
     impl/chain_spec_impl.cpp

core/application/app_configuration.hpp

@@ -252,6 +252,8 @@ namespace kagome::application {
      */
     virtual StorageBackend storageBackend() const = 0;
 
+    virtual std::optional<size_t> statePruningDepth() const = 0;
+
     /**
      * @return database state cache size in MiB
      */

core/application/impl/app_configuration_impl.cpp

@@ -5,7 +5,6 @@
 
 #include "application/impl/app_configuration_impl.hpp"
 
-#include <fstream>
 #include <limits>
 #include <regex>
 #include <string>
@@ -18,7 +17,6 @@
 #include <boost/uuid/uuid_io.hpp>
 #include <charconv>
 #include <libp2p/layer/websocket/wss_adaptor.hpp>
-#include "filesystem/common.hpp"
 
 #include "api/transport/tuner.hpp"
 #include "application/build_version.hpp"
@@ -27,6 +25,7 @@
 #include "chain_spec_impl.hpp"
 #include "common/hexutil.hpp"
 #include "common/uri.hpp"
+#include "filesystem/common.hpp"
 #include "filesystem/directories.hpp"
 #include "utils/read_file.hpp"
 
@@ -248,7 +247,8 @@ namespace kagome::application {
         offchain_worker_mode_{def_offchain_worker_mode},
         enable_offchain_indexing_{def_enable_offchain_indexing},
         recovery_state_{def_block_to_recover},
-        db_cache_size_{def_db_cache_size} {
+        db_cache_size_{def_db_cache_size},
+        state_pruning_depth_{} {
     SL_INFO(logger_, "Soramitsu Kagome started. Version: {} ", buildVersion());
   }
 
@@ -790,6 +790,7 @@ namespace kagome::application {
         ("db-cache", po::value<uint32_t>()->default_value(def_db_cache_size), "Limit the memory the database cache can use <MiB>")
         ("enable-offchain-indexing", po::value<bool>(), "enable Offchain Indexing API, which allow block import to write to offchain DB)")
         ("recovery", po::value<std::string>(), "recovers block storage to state after provided block presented by number or hash, and stop after that")
+        ("state-pruning", po::value<std::string>()->default_value("archive"), "state pruning policy. 'archive' or the number of finalized blocks to keep.")
         ;
 
     po::options_description network_desc("Network options");
@@ -1445,6 +1446,27 @@ namespace kagome::application {
       return false;
     }
 
+    if (auto state_pruning_opt =
+            find_argument<std::string>(vm, "state-pruning");
+        state_pruning_opt.has_value()) {
+      const auto& val = state_pruning_opt.value();
+      if (val == "archive") {
+        state_pruning_depth_ = std::nullopt;
+      } else {
+        uint32_t depth{};
+        auto [_, err] = std::from_chars(&*val.begin(), &*val.end(), depth);
+        if (err == std::errc{}) {
+          state_pruning_depth_ = depth;
+        } else {
+          SL_ERROR(logger_,
+                   "Failed to parse state-pruning param (which should be "
+                   "either 'archive' or an integer): {}",
+                   err);
+          return false;
+        }
+      }
+    }
+
     // if something wrong with config print help message
     if (not validate_config()) {
       std::cout << desc << std::endl;

core/application/impl/app_configuration_impl.hpp

@@ -184,6 +184,9 @@ namespace kagome::application {
     uint32_t dbCacheSize() const override {
       return db_cache_size_;
     }
+    std::optional<size_t> statePruningDepth() const override {
+      return state_pruning_depth_;
+    }
     std::optional<std::string_view> devMnemonicPhrase() const override {
       if (dev_mnemonic_phrase_) {
         return *dev_mnemonic_phrase_;
@@ -337,6 +340,7 @@ namespace kagome::application {
     std::optional<primitives::BlockId> recovery_state_;
     StorageBackend storage_backend_ = StorageBackend::RocksDB;
     uint32_t db_cache_size_;
+    std::optional<size_t> state_pruning_depth_;
     std::optional<std::string> dev_mnemonic_phrase_;
     std::string node_wss_pem_;
     std::optional<BenchmarkConfigSection> benchmark_config_;

core/application/impl/chain_spec_impl.cpp

@@ -254,10 +254,10 @@ namespace kagome::application {
 
     auto read_key_block = [](const auto &tree,
                              GenesisRawData &data) -> outcome::result<void> {
-      for (const auto &[child_key, child_value] : tree) {
+      for (const auto &[key, value] : tree) {
         // get rid of leading 0x for key and value and unhex
-        OUTCOME_TRY(key_processed, common::unhexWith0x(child_key));
-        OUTCOME_TRY(value_processed, common::unhexWith0x(child_value.data()));
+        OUTCOME_TRY(key_processed, common::unhexWith0x(key));
+        OUTCOME_TRY(value_processed, common::unhexWith0x(value.data()));
         data.emplace_back(std::move(key_processed), std::move(value_processed));
       }
       return outcome::success();

core/authority_discovery/query/query_impl.cpp

@@ -122,7 +122,7 @@ namespace kagome::authority_discovery {
       auto authority = queue_.back();
       queue_.pop_back();
 
-      common::Buffer hash = crypto::sha256(authority);
+      common::Buffer hash{crypto::sha256(authority)};
       scheduler_->schedule([=, wp = weak_from_this()] {
         if (auto self = wp.lock()) {
           std::ignore = kademlia_->getValue(
@@ -131,12 +131,12 @@ namespace kagome::authority_discovery {
                 --active_;
                 pop();
                 if (res.has_error()) {
-                  SL_WARN(log_, "Kademlia can't get value: {}", res.error());
+                  SL_DEBUG(log_, "Kademlia can't get value: {}", res.error());
                   return;
                 }
                 auto r = add(authority, std::move(res.value()));
                 if (not r) {
-                  SL_WARN(log_, "Can't add: {}", r.error());
+                  SL_DEBUG(log_, "Can't add: {}", r.error());
                 }
               });
         }

core/benchmark/block_execution_benchmark.cpp

@@ -20,6 +20,8 @@ OUTCOME_CPP_DEFINE_CATEGORY(kagome::benchmark,
     using E = kagome::benchmark::BlockExecutionBenchmark::Error;
     case E::BLOCK_WEIGHT_DECODE_FAILED:
       return "Failed to decode block weight";
+    case E::BLOCK_NOT_FOUND:
+      return "A block expected to be present in the block tree is not found";
   }
   return "Unknown BlockExecutionBenchmark error";
 }
@@ -221,8 +223,12 @@ namespace kagome::benchmark {
                     block_tree_->getBlockHash(config.start),
                     "retrieving hash of block {}",
                     config.start);
+    if (!current_hash) {
+      SL_ERROR(logger_, "Start block {} is not found!", config.start);
+      return Error::BLOCK_NOT_FOUND;
+    }
 
-    primitives::BlockInfo current_block_info = {config.start, current_hash};
+    primitives::BlockInfo current_block_info = {config.start, *current_hash};
     std::vector<primitives::BlockHash> block_hashes;
     std::vector<primitives::Block> blocks;
     while (current_block_info.number <= config.end) {
@@ -244,7 +250,14 @@ namespace kagome::benchmark {
                       "retrieving hash of block {}",
                       current_block_info.number + 1);
       current_block_info.number += 1;
-      current_block_info.hash = next_hash;
+
+      if (!next_hash) {
+        SL_ERROR(logger_,
+                 "Next block {} is not found!",
+                 current_block_info.number + 1);
+        return Error::BLOCK_NOT_FOUND;
+      }
+      current_block_info.hash = *next_hash;
     }
 
     std::chrono::steady_clock clock;
@@ -291,7 +304,8 @@ namespace kagome::benchmark {
               *trie_storage_,
               blocks[stat.getBlock().number - config.start].header.state_root));
       fmt::print(
-          "Block #{}: consumed {} ns out of declared {} ns on average. ({} %)\n",
+          "Block #{}: consumed {} ns out of declared {} ns on average. ({} "
+          "%)\n",
           stat.getBlock().number,
           stat.avg().count(),
           block_weight_ns.count(),

core/benchmark/block_execution_benchmark.hpp

@@ -32,6 +32,7 @@ namespace kagome::benchmark {
    public:
     enum class Error {
       BLOCK_WEIGHT_DECODE_FAILED,
+      BLOCK_NOT_FOUND,
     };
 
     struct Config {

core/blockchain/block_tree.hpp

@@ -43,7 +43,7 @@ namespace kagome::blockchain {
      * @param block_number of the block header we are looking for
      * @return result containing block hash if it exists, error otherwise
      */
-    virtual outcome::result<primitives::BlockHash> getBlockHash(
+    virtual outcome::result<std::optional<primitives::BlockHash>> getBlockHash(
         primitives::BlockNumber block_number) const = 0;
 
     /**