Changes tracker

core/api/service/impl/api_service_impl.cpp

@@ -284,17 +284,17 @@ namespace kagome::api {
                 const auto &header =
                     block_tree_->getBlockHeader(best_block_hash);
                 BOOST_ASSERT(header.has_value());
-                auto persistent_batch = trie_storage_->getPersistentBatchAt(
+                auto batch_res = trie_storage_->getEphemeralBatchAt(
                     header.value().state_root);
-                if (!persistent_batch.has_value()) {
+                if (!batch_res.has_value()) {
                   SL_ERROR(logger_,
                            "Failed to get storage state for block {}, required "
                            "to subscribe an RPC session to some storage keys.",
                            best_block_hash);
-                  return persistent_batch.as_failure();
+                  return batch_res.as_failure();
                 }
 
-                auto &batch = persistent_batch.value();
+                auto &batch = batch_res.value();
 
                 session_context.messages = uploadMessagesListFromCache();
 

core/authorship/block_builder_factory.hpp

@@ -29,7 +29,8 @@ namespace kagome::authorship {
      */
     virtual outcome::result<std::unique_ptr<BlockBuilder>> make(
         const primitives::BlockInfo &parent_block,
-        primitives::Digest inherent_digest) const = 0;
+        primitives::Digest inherent_digest,
+        TrieChangesTrackerOpt changes_tracker) const = 0;
   };
 
 }  // namespace kagome::authorship

core/authorship/impl/block_builder_factory_impl.cpp

@@ -24,7 +24,8 @@ namespace kagome::authorship {
 
   outcome::result<std::unique_ptr<BlockBuilder>> BlockBuilderFactoryImpl::make(
       const kagome::primitives::BlockInfo &parent,
-      primitives::Digest inherent_digest) const {
+      primitives::Digest inherent_digest,
+      TrieChangesTrackerOpt changes_tracker) const {
 #ifndef BOOST_ASSERT_IS_VOID
     OUTCOME_TRY(parent_number, header_backend_->getNumberById(parent.hash));
 #endif
@@ -36,7 +37,9 @@ namespace kagome::authorship {
     header.parent_hash = parent.hash;
     header.digest = std::move(inherent_digest);
 
-    if (auto res = r_core_->initialize_block(header); not res) {
+    if (auto res =
+            r_core_->initialize_block(header, std::move(changes_tracker));
+        not res) {
       logger_->error("Core_initialize_block failed: {}", res.error());
       return res.error();
     } else {

core/authorship/impl/block_builder_factory_impl.hpp

@@ -24,7 +24,8 @@ namespace kagome::authorship {
 
     outcome::result<std::unique_ptr<BlockBuilder>> make(
         const kagome::primitives::BlockInfo &parent_block,
-        primitives::Digest inherent_digest) const override;
+        primitives::Digest inherent_digest,
+        TrieChangesTrackerOpt changes_tracker) const override;
 
    private:
     std::shared_ptr<runtime::Core> r_core_;

core/authorship/impl/proposer_impl.cpp

@@ -40,9 +40,11 @@ namespace kagome::authorship {
   outcome::result<primitives::Block> ProposerImpl::propose(
       const primitives::BlockInfo &parent_block,
       const primitives::InherentData &inherent_data,
-      const primitives::Digest &inherent_digest) {
+      const primitives::Digest &inherent_digest,
+      TrieChangesTrackerOpt changes_tracker) {
     OUTCOME_TRY(block_builder,
-                block_builder_factory_->make(parent_block, inherent_digest));
+                block_builder_factory_->make(
+                    parent_block, inherent_digest, std::move(changes_tracker)));
 
     auto inherent_xts_res = block_builder->getInherentExtrinsics(inherent_data);
     if (not inherent_xts_res) {

core/authorship/impl/proposer_impl.hpp

@@ -39,7 +39,8 @@ namespace kagome::authorship {
     outcome::result<primitives::Block> propose(
         const primitives::BlockInfo &parent_block,
         const primitives::InherentData &inherent_data,
-        const primitives::Digest &inherent_digest) override;
+        const primitives::Digest &inherent_digest,
+        TrieChangesTrackerOpt changes_tracker) override;
 
    private:
     std::shared_ptr<BlockBuilderFactory> block_builder_factory_;

core/authorship/proposer.hpp

@@ -11,6 +11,7 @@
 #include "primitives/common.hpp"
 #include "primitives/digest.hpp"
 #include "primitives/inherent_data.hpp"
+#include "storage/changes_trie/changes_tracker.hpp"
 
 namespace kagome::authorship {
 
@@ -31,7 +32,8 @@ namespace kagome::authorship {
     virtual outcome::result<primitives::Block> propose(
         const primitives::BlockInfo &parent_block,
         const primitives::InherentData &inherent_data,
-        const primitives::Digest &inherent_digest) = 0;
+        const primitives::Digest &inherent_digest,
+        TrieChangesTrackerOpt changes_tracker) = 0;
   };
 
 }  // namespace kagome::authorship

core/blockchain/impl/block_tree_impl.cpp

@@ -16,7 +16,6 @@
 #include "consensus/babe/is_primary.hpp"
 #include "crypto/blake2/blake2b.h"
 #include "log/profiling_logger.hpp"
-#include "storage/changes_trie/changes_tracker.hpp"
 #include "storage/database_error.hpp"
 
 namespace {
@@ -116,7 +115,6 @@ namespace kagome::blockchain {
           extrinsic_events_engine,
       std::shared_ptr<subscription::ExtrinsicEventKeyRepository>
           extrinsic_event_key_repo,
-      std::shared_ptr<storage::changes_trie::ChangesTracker> changes_tracker,
       std::shared_ptr<const class JustificationStoragePolicy>
           justification_storage_policy) {
     BOOST_ASSERT(storage != nullptr);
@@ -213,7 +211,6 @@ namespace kagome::blockchain {
                           std::move(chain_events_engine),
                           std::move(extrinsic_events_engine),
                           std::move(extrinsic_event_key_repo),
-                          std::move(changes_tracker),
                           std::move(justification_storage_policy)));
 
     // Add non-finalized block to the block tree
@@ -344,7 +341,6 @@ namespace kagome::blockchain {
           extrinsic_events_engine,
       std::shared_ptr<subscription::ExtrinsicEventKeyRepository>
           extrinsic_event_key_repo,
-      std::shared_ptr<storage::changes_trie::ChangesTracker> changes_tracker,
       std::shared_ptr<const JustificationStoragePolicy>
           justification_storage_policy)
       : header_repo_{std::move(header_repo)},
@@ -355,7 +351,6 @@ namespace kagome::blockchain {
         chain_events_engine_(std::move(chain_events_engine)),
         extrinsic_events_engine_(std::move(extrinsic_events_engine)),
         extrinsic_event_key_repo_{std::move(extrinsic_event_key_repo)},
-        trie_changes_tracker_(std::move(changes_tracker)),
         justification_storage_policy_{std::move(justification_storage_policy)} {
     BOOST_ASSERT(header_repo_ != nullptr);
     BOOST_ASSERT(storage_ != nullptr);
@@ -365,7 +360,6 @@ namespace kagome::blockchain {
     BOOST_ASSERT(chain_events_engine_ != nullptr);
     BOOST_ASSERT(extrinsic_events_engine_ != nullptr);
     BOOST_ASSERT(extrinsic_event_key_repo_ != nullptr);
-    BOOST_ASSERT(trie_changes_tracker_ != nullptr);
     BOOST_ASSERT(justification_storage_policy_ != nullptr);
     BOOST_ASSERT(telemetry_ != nullptr);
 
@@ -470,7 +464,6 @@ namespace kagome::blockchain {
 
     chain_events_engine_->notify(primitives::events::ChainEventType::kNewHeads,
                                  block.header);
-    trie_changes_tracker_->onBlockAdded(block_hash);
     SL_DEBUG(log_, "Adding block {}", block_hash);
     for (const auto &ext : block.body) {
       auto hash = hasher_->blake2b_256(ext.data);

core/blockchain/impl/block_tree_impl.hpp

@@ -31,10 +31,6 @@
 #include "subscription/extrinsic_event_key_repository.hpp"
 #include "telemetry/service.hpp"
 
-namespace kagome::storage::changes_trie {
-  class ChangesTracker;
-}
-
 namespace kagome::blockchain {
 
   class TreeNode;
@@ -54,7 +50,6 @@ namespace kagome::blockchain {
             extrinsic_events_engine,
         std::shared_ptr<subscription::ExtrinsicEventKeyRepository>
             extrinsic_event_key_repo,
-        std::shared_ptr<storage::changes_trie::ChangesTracker> changes_tracker,
         std::shared_ptr<const class JustificationStoragePolicy>
             justification_storage_policy);
 
@@ -151,7 +146,6 @@ namespace kagome::blockchain {
             extrinsic_events_engine,
         std::shared_ptr<subscription::ExtrinsicEventKeyRepository>
             extrinsic_event_key_repo,
-        std::shared_ptr<storage::changes_trie::ChangesTracker> changes_tracker,
         std::shared_ptr<const class JustificationStoragePolicy>
             justification_storage_policy);
 
@@ -185,8 +179,6 @@ namespace kagome::blockchain {
     primitives::events::ExtrinsicSubscriptionEnginePtr extrinsic_events_engine_;
     std::shared_ptr<subscription::ExtrinsicEventKeyRepository>
         extrinsic_event_key_repo_;
-    std::shared_ptr<storage::changes_trie::ChangesTracker>
-        trie_changes_tracker_;
     std::shared_ptr<const class JustificationStoragePolicy>
         justification_storage_policy_;
 

core/consensus/babe/impl/babe_impl.cpp

@@ -27,6 +27,7 @@
 #include "network/types/collator_messages.hpp"
 #include "runtime/runtime_api/core.hpp"
 #include "runtime/runtime_api/offchain_worker_api.hpp"
+#include "storage/changes_trie/impl/storage_changes_tracker_impl.hpp"
 #include "storage/trie/serialization/ordered_trie_hash.hpp"
 #include "storage/trie/trie_storage.hpp"
 
@@ -60,6 +61,7 @@ namespace kagome::consensus::babe {
       std::shared_ptr<BabeUtil> babe_util,
       std::shared_ptr<parachain::BitfieldStore> bitfield_store,
       std::shared_ptr<parachain::BackingStore> backing_store,
+      primitives::events::StorageSubscriptionEnginePtr storage_sub_engine,
       primitives::events::ChainSubscriptionEnginePtr chain_events_engine,
       std::shared_ptr<runtime::OffchainWorkerApi> offchain_worker_api,
       std::shared_ptr<runtime::Core> core,
@@ -83,6 +85,7 @@ namespace kagome::consensus::babe {
         babe_util_(std::move(babe_util)),
         bitfield_store_{std::move(bitfield_store)},
         backing_store_{std::move(backing_store)},
+        storage_sub_engine_{std::move(storage_sub_engine)},
         chain_events_engine_(std::move(chain_events_engine)),
         chain_sub_([&] {
           BOOST_ASSERT(chain_events_engine_ != nullptr);
@@ -972,9 +975,12 @@ namespace kagome::consensus::babe {
     }
     const auto &babe_pre_digest = babe_pre_digest_res.value();
 
+    auto changes_tracker =
+        std::make_shared<storage::changes_trie::StorageChangesTrackerImpl>();
+
     // create new block
-    auto pre_seal_block_res =
-        proposer_->propose(best_block_, inherent_data, {babe_pre_digest});
+    auto pre_seal_block_res = proposer_->propose(
+        best_block_, inherent_data, {babe_pre_digest}, changes_tracker);
     if (!pre_seal_block_res) {
       SL_ERROR(log_, "Cannot propose a block: {}", pre_seal_block_res.error());
       return;
@@ -1053,6 +1059,10 @@ namespace kagome::consensus::babe {
       }
       return;
     }
+
+    changes_tracker->onBlockAdded(
+        block_hash, storage_sub_engine_, chain_events_engine_);
+
     telemetry_->notifyBlockImported(block_info, telemetry::BlockOrigin::kOwn);
 
     // observe digest of block

core/consensus/babe/impl/babe_impl.hpp

@@ -86,31 +86,33 @@ namespace kagome::consensus::babe {
     /**
      * Create an instance of Babe implementation
      */
-    BabeImpl(const application::AppConfiguration &app_config,
-             std::shared_ptr<application::AppStateManager> app_state_manager,
-             std::shared_ptr<BabeLottery> lottery,
-             std::shared_ptr<BabeConfigRepository> babe_config_repo,
-             std::shared_ptr<authorship::Proposer> proposer,
-             std::shared_ptr<blockchain::BlockTree> block_tree,
-             std::shared_ptr<network::BlockAnnounceTransmitter>
-                 block_announce_transmitter,
-             std::shared_ptr<crypto::Sr25519Provider> sr25519_provider,
-             std::shared_ptr<crypto::Sr25519Keypair> keypair,
-             std::shared_ptr<clock::SystemClock> clock,
-             std::shared_ptr<crypto::Hasher> hasher,
-             std::unique_ptr<clock::Timer> timer,
-             std::shared_ptr<blockchain::DigestTracker> digest_tracker,
-             std::shared_ptr<network::Synchronizer> synchronizer,
-             std::shared_ptr<BabeUtil> babe_util,
-             std::shared_ptr<parachain::BitfieldStore> bitfield_store,
-             std::shared_ptr<parachain::BackingStore> backing_store,
-             primitives::events::ChainSubscriptionEnginePtr chain_events_engine,
-             std::shared_ptr<runtime::OffchainWorkerApi> offchain_worker_api,
-             std::shared_ptr<runtime::Core> core,
-             std::shared_ptr<ConsistencyKeeper> consistency_keeper,
-             std::shared_ptr<storage::trie::TrieStorage> trie_storage,
-             primitives::events::BabeStateSubscriptionEnginePtr
-                 babe_status_observable);
+    BabeImpl(
+        const application::AppConfiguration &app_config,
+        std::shared_ptr<application::AppStateManager> app_state_manager,
+        std::shared_ptr<BabeLottery> lottery,
+        std::shared_ptr<BabeConfigRepository> babe_config_repo,
+        std::shared_ptr<authorship::Proposer> proposer,
+        std::shared_ptr<blockchain::BlockTree> block_tree,
+        std::shared_ptr<network::BlockAnnounceTransmitter>
+            block_announce_transmitter,
+        std::shared_ptr<crypto::Sr25519Provider> sr25519_provider,
+        std::shared_ptr<crypto::Sr25519Keypair> keypair,
+        std::shared_ptr<clock::SystemClock> clock,
+        std::shared_ptr<crypto::Hasher> hasher,
+        std::unique_ptr<clock::Timer> timer,
+        std::shared_ptr<blockchain::DigestTracker> digest_tracker,
+        std::shared_ptr<network::Synchronizer> synchronizer,
+        std::shared_ptr<BabeUtil> babe_util,
+        std::shared_ptr<parachain::BitfieldStore> bitfield_store,
+        std::shared_ptr<parachain::BackingStore> backing_store,
+        primitives::events::StorageSubscriptionEnginePtr storage_sub_engine,
+        primitives::events::ChainSubscriptionEnginePtr chain_events_engine,
+        std::shared_ptr<runtime::OffchainWorkerApi> offchain_worker_api,
+        std::shared_ptr<runtime::Core> core,
+        std::shared_ptr<ConsistencyKeeper> consistency_keeper,
+        std::shared_ptr<storage::trie::TrieStorage> trie_storage,
+        primitives::events::BabeStateSubscriptionEnginePtr
+            babe_status_observable);
 
     ~BabeImpl() override = default;
 
@@ -201,6 +203,7 @@ namespace kagome::consensus::babe {
     std::shared_ptr<BabeUtil> babe_util_;
     std::shared_ptr<parachain::BitfieldStore> bitfield_store_;
     std::shared_ptr<parachain::BackingStore> backing_store_;
+    primitives::events::StorageSubscriptionEnginePtr storage_sub_engine_;
     primitives::events::ChainSubscriptionEnginePtr chain_events_engine_;
     std::shared_ptr<primitives::events::ChainEventSubscriber> chain_sub_;
     std::optional<primitives::Version> actual_runtime_version_;

core/consensus/babe/impl/block_executor_impl.cpp

@@ -18,6 +18,7 @@
 #include "consensus/validation/block_validator.hpp"
 #include "runtime/runtime_api/core.hpp"
 #include "runtime/runtime_api/offchain_worker_api.hpp"
+#include "storage/changes_trie/impl/storage_changes_tracker_impl.hpp"
 #include "transaction_pool/transaction_pool.hpp"
 #include "transaction_pool/transaction_pool_error.hpp"
 
@@ -34,12 +35,16 @@ namespace kagome::consensus::babe {
       std::shared_ptr<transaction_pool::TransactionPool> tx_pool,
       std::shared_ptr<crypto::Hasher> hasher,
       std::shared_ptr<runtime::OffchainWorkerApi> offchain_worker_api,
+      primitives::events::StorageSubscriptionEnginePtr storage_sub_engine,
+      primitives::events::ChainSubscriptionEnginePtr chain_sub_engine,
       std::unique_ptr<BlockAppenderBase> appender)
       : block_tree_{std::move(block_tree)},
         core_{std::move(core)},
         tx_pool_{std::move(tx_pool)},
         hasher_{std::move(hasher)},
         offchain_worker_api_(std::move(offchain_worker_api)),
+        storage_sub_engine_{std::move(storage_sub_engine)},
+        chain_subscription_engine_{std::move(chain_sub_engine)},
         appender_{std::move(appender)},
         logger_{log::createLogger("BlockExecutor", "block_executor")},
         telemetry_{telemetry::createTelemetryService()} {
@@ -123,7 +128,11 @@ namespace kagome::consensus::babe {
       // block should be applied without last digest which contains the seal
       block_without_seal_digest.header.digest.pop_back();
 
-      OUTCOME_TRY(core_->execute_block(block_without_seal_digest));
+      auto changes_tracker =
+          std::make_shared<storage::changes_trie::StorageChangesTrackerImpl>();
+
+      OUTCOME_TRY(
+          core_->execute_block(block_without_seal_digest, changes_tracker));
 
       auto exec_end = std::chrono::high_resolution_clock::now();
       auto duration_ms = std::chrono::duration_cast<std::chrono::milliseconds>(
@@ -136,6 +145,9 @@ namespace kagome::consensus::babe {
 
       // add block header if it does not exist
       OUTCOME_TRY(block_tree_->addBlock(block));
+
+      changes_tracker->onBlockAdded(
+          block_info.hash, storage_sub_engine_, chain_subscription_engine_);
     }
 
     OUTCOME_TRY(appender_->applyJustifications(block_info, justification));
@@ -154,9 +166,8 @@ namespace kagome::consensus::babe {
     }
 
     OUTCOME_TRY(slot_info, appender_->getSlotInfo(block.header));
-    auto& [slot_start, slot_duration] = slot_info;
-    auto lag = std::chrono::system_clock::now()
-             - slot_start;
+    auto &[slot_start, slot_duration] = slot_info;
+    auto lag = std::chrono::system_clock::now() - slot_start;
     std::string lag_msg;
     if (lag > std::chrono::hours(99)) {
       lag_msg = fmt::format(

core/consensus/babe/impl/block_executor_impl.hpp

@@ -14,6 +14,7 @@
 #include "metrics/metrics.hpp"
 #include "primitives/babe_configuration.hpp"
 #include "primitives/block_header.hpp"
+#include "primitives/event_types.hpp"
 #include "telemetry/service.hpp"
 
 namespace kagome::runtime {
@@ -47,6 +48,8 @@ namespace kagome::consensus::babe {
         std::shared_ptr<transaction_pool::TransactionPool> tx_pool,
         std::shared_ptr<crypto::Hasher> hasher,
         std::shared_ptr<runtime::OffchainWorkerApi> offchain_worker_api,
+        primitives::events::StorageSubscriptionEnginePtr storage_sub_engine,
+        primitives::events::ChainSubscriptionEnginePtr chain_sub_engine,
         std::unique_ptr<BlockAppenderBase> appender);
 
     ~BlockExecutorImpl();
@@ -61,6 +64,8 @@ namespace kagome::consensus::babe {
     std::shared_ptr<transaction_pool::TransactionPool> tx_pool_;
     std::shared_ptr<crypto::Hasher> hasher_;
     std::shared_ptr<runtime::OffchainWorkerApi> offchain_worker_api_;
+    primitives::events::StorageSubscriptionEnginePtr storage_sub_engine_;
+    primitives::events::ChainSubscriptionEnginePtr chain_subscription_engine_;
 
     std::unique_ptr<BlockAppenderBase> appender_;