feature/Ellastic-scaling + priority nodes + cluster integr

cmake/Hunter/init.cmake

@@ -26,5 +26,11 @@ set(HUNTER_CACHE_SERVERS
     CACHE STRING "Binary cache server"
 )
 
+# https://hunter.readthedocs.io/en/latest/reference/user-variables.html#hunter-use-cache-servers
+# set(
+#     HUNTER_USE_CACHE_SERVERS NO
+#     CACHE STRING "Disable binary cache"
+# )
+
 include(${CMAKE_CURRENT_LIST_DIR}/HunterGate.cmake)
 include(${CMAKE_CURRENT_LIST_DIR}/hunter-gate-url.cmake)

core/api/jrpc/value_converter.hpp

@@ -70,6 +70,10 @@ namespace kagome::api {
   inline jsonrpc::Value makeValue(const primitives::Version &);
   inline jsonrpc::Value makeValue(const primitives::Justification &);
   inline jsonrpc::Value makeValue(const primitives::RpcMethods &);
+  inline jsonrpc::Value makeValue(
+      const primitives::events::RemoveAfterFinalizationParams &val);
+  inline jsonrpc::Value makeValue(
+      const primitives::events::RemoveAfterFinalizationParams::HeaderInfo &val);
 
   inline jsonrpc::Value makeValue(const uint32_t &val) {
     return static_cast<int64_t>(val);
@@ -150,6 +154,17 @@ namespace kagome::api {
     return value;
   }
 
+  inline jsonrpc::Value makeValue(
+      const primitives::events::RemoveAfterFinalizationParams &val) {
+    return makeValue(val.removed);
+  }
+
+  inline jsonrpc::Value makeValue(
+      const primitives::events::RemoveAfterFinalizationParams::HeaderInfo
+          &val) {
+    return makeValue(val.hash);
+  }
+
   template <size_t N>
   inline jsonrpc::Value makeValue(const common::Blob<N> &val) {
     return makeValue(BufferView{val});

core/blockchain/impl/block_tree_impl.cpp

@@ -813,9 +813,13 @@ namespace kagome::blockchain {
 
         OUTCOME_TRY(p.storage_->putJustification(justification, block_hash));
 
-        std::vector<primitives::BlockHash> retired_hashes;
+        std::vector<
+            primitives::events::RemoveAfterFinalizationParams::HeaderInfo>
+            retired_hashes;
         for (auto parent = node->parent(); parent; parent = parent->parent()) {
-          retired_hashes.emplace_back(parent->info.hash);
+          retired_hashes.emplace_back(
+              primitives::events::RemoveAfterFinalizationParams::HeaderInfo{
+                  parent->info.hash, parent->info.number});
         }
 
         auto changes = p.tree_->finalize(node);
@@ -846,12 +850,13 @@ namespace kagome::blockchain {
 
         main_pool_handler_->execute(
             [weak{weak_from_this()},
-             retired_hashes{std::move(retired_hashes)}] {
+             retired{primitives::events::RemoveAfterFinalizationParams{
+                 std::move(retired_hashes), header.number}}] {
               if (auto self = weak.lock()) {
                 self->chain_events_engine_->notify(
                     primitives::events::ChainEventType::
                         kDeactivateAfterFinalization,
-                    retired_hashes);
+                    retired);
               }
             });
 
@@ -1301,7 +1306,8 @@ namespace kagome::blockchain {
     }
 
     std::vector<primitives::Extrinsic> extrinsics;
-    std::vector<primitives::BlockHash> retired_hashes;
+    std::vector<primitives::events::RemoveAfterFinalizationParams::HeaderInfo>
+        retired_hashes;
 
     // remove from storage
     retired_hashes.reserve(changes.prune.size());
@@ -1329,33 +1335,39 @@ namespace kagome::blockchain {
         BOOST_ASSERT(block_header_opt.has_value());
         OUTCOME_TRY(p.state_pruner_->pruneDiscarded(block_header_opt.value()));
       }
-      retired_hashes.emplace_back(block.hash);
+      retired_hashes.emplace_back(
+          primitives::events::RemoveAfterFinalizationParams::HeaderInfo{
+              block.hash, block.number});
       OUTCOME_TRY(p.storage_->removeBlock(block.hash));
     }
 
     // trying to return extrinsics back to transaction pool
-    main_pool_handler_->execute([extrinsics{std::move(extrinsics)},
-                                 wself{weak_from_this()},
-                                 retired_hashes{
-                                     std::move(retired_hashes)}]() mutable {
-      if (auto self = wself.lock()) {
-        auto eo = self->block_tree_data_.sharedAccess(
-            [&](const BlockTreeData &p) { return p.extrinsic_observer_; });
-
-        for (auto &&extrinsic : extrinsics) {
-          auto result = eo->onTxMessage(extrinsic);
-          if (result) {
-            SL_DEBUG(self->log_, "Tx {} was reapplied", result.value().toHex());
-          } else {
-            SL_DEBUG(self->log_, "Tx was skipped: {}", result.error());
-          }
-        }
+    main_pool_handler_->execute(
+        [extrinsics{std::move(extrinsics)},
+         wself{weak_from_this()},
+         retired{primitives::events::RemoveAfterFinalizationParams{
+             std::move(retired_hashes),
+             getLastFinalizedNoLock(p).number}}]() mutable {
+          if (auto self = wself.lock()) {
+            auto eo = self->block_tree_data_.sharedAccess(
+                [&](const BlockTreeData &p) { return p.extrinsic_observer_; });
+
+            for (auto &&extrinsic : extrinsics) {
+              auto result = eo->onTxMessage(extrinsic);
+              if (result) {
+                SL_DEBUG(
+                    self->log_, "Tx {} was reapplied", result.value().toHex());
+              } else {
+                SL_DEBUG(self->log_, "Tx was skipped: {}", result.error());
+              }
+            }
 
-        self->chain_events_engine_->notify(
-            primitives::events::ChainEventType::kDeactivateAfterFinalization,
-            retired_hashes);
-      }
-    });
+            self->chain_events_engine_->notify(
+                primitives::events::ChainEventType::
+                    kDeactivateAfterFinalization,
+                retired);
+          }
+        });
 
     return outcome::success();
   }

core/dispute_coordinator/impl/dispute_coordinator_impl.cpp

@@ -63,7 +63,7 @@ namespace kagome::dispute {
           [&](const ValidDisputeStatement &kind) {
             return visit_in_place(
                 kind,
-                [&](const Explicit &) {
+                [&](const Explicit &x) {
                   std::array<uint8_t, 4> magic{'D', 'I', 'S', 'P'};
                   bool validity = true;
                   return scale::encode(
@@ -91,6 +91,15 @@ namespace kagome::dispute {
                   std::array<uint8_t, 4> magic{'A', 'P', 'P', 'R'};
                   return scale::encode(
                       std::tie(magic, candidate_hash, session));
+                },
+                [&](const ApprovalCheckingMultipleCandidates &candidates) {
+                  std::array<uint8_t, 4> magic{'A', 'P', 'P', 'R'};
+                  if (candidates.size() == 1) {
+                    return scale::encode(
+                        std::tie(magic, candidates.front(), session));
+                  } else {
+                    return scale::encode(std::tie(magic, candidates, session));
+                  }
                 });
           },
           [&](const InvalidDisputeStatement &kind) {
@@ -320,7 +329,7 @@ namespace kagome::dispute {
     // https://github.com/paritytech/polkadot/blob/40974fb99c86f5c341105b7db53c7aa0df707d66/node/core/dispute-coordinator/src/lib.rs#L298
     for (auto &[session, candidate_hash, status] : active_disputes) {
       auto env_opt = makeCandidateEnvironment(
-          *session_keys_, *rolling_session_window_, session);
+          *session_keys_, *rolling_session_window_, session, first_leaf.hash);
       if (not env_opt.has_value()) {
         continue;
       }
@@ -535,6 +544,7 @@ namespace kagome::dispute {
     auto res = process_active_leaves_update(update);
     if (res.has_error()) {
       SL_ERROR(log_, "Can't handle active list update: {}", res.error());
+      std::ignore = process_active_leaves_update(update);
       return;
     }
   }
@@ -550,18 +560,6 @@ namespace kagome::dispute {
       return outcome::success();
     }
 
-    // Obtain the session info, for sake of `ValidatorId`s either from the
-    // rolling session window. Must be called _after_ `fn
-    // cache_session_info_for_head` which guarantees that the session info is
-    // available for the current session.
-    auto session_info_opt = rolling_session_window_->session_info(session);
-    if (not session_info_opt.has_value()) {
-      SL_WARN(log_,
-              "Could not retrieve session info from rolling session window");
-      return outcome::success();
-    }
-    auto &session_info = session_info_opt.value().get();
-
     // Scraped on-chain backing votes for the candidates with the new active
     // leaf as if we received them via gossip.
     for (auto &[candidate_receipt, backers] :
@@ -571,6 +569,11 @@ namespace kagome::dispute {
 
       SL_TRACE(log_, "Importing backing votes from chain for candidate");
 
+      OUTCOME_TRY(session, api_->session_index_for_child(relay_parent));
+      OUTCOME_TRY(session_info_opt, api_->session_info(relay_parent, session));
+      BOOST_ASSERT(session_info_opt.has_value());
+      const auto &session_info = session_info_opt.value();
+
       std::vector<Indexed<SignedDisputeStatement>> statements;
       for (auto &[validator_index, attestation] : backers) {
         if (validator_index >= session_info.validators.size()) {
@@ -978,12 +981,18 @@ namespace kagome::dispute {
   DisputeCoordinatorImpl::makeCandidateEnvironment(
       crypto::SessionKeys &session_keys,
       RollingSessionWindow &rolling_session_window,
-      SessionIndex session) {
-    auto session_info_opt = rolling_session_window.session_info(session);
+      SessionIndex session,
+      primitives::BlockHash relay_parent) {
+    auto session_info_opt_res = api_->session_info(relay_parent, session);
+    if (session_info_opt_res.has_error()) {
+      // TODO log
+      return std::nullopt;
+    }
+    auto session_info_opt = session_info_opt_res.value();
     if (not session_info_opt.has_value()) {
       return std::nullopt;
     }
-    auto &session_info = session_info_opt.value().get();
+    auto session_info = std::move(session_info_opt.value());
 
     std::unordered_set<ValidatorIndex> controlled_indices;
     auto keypair = session_keys.getParaKeyPair(session_info.validators);
@@ -992,7 +1001,7 @@ namespace kagome::dispute {
     }
 
     return CandidateEnvironment{.session_index = session,
-                                .session = session_info,
+                                .session = std::move(session_info),
                                 .controlled_indices = controlled_indices};
   }
 
@@ -1015,16 +1024,6 @@ namespace kagome::dispute {
                   .hash(*hasher_)
             : boost::relaxed_get<CandidateHash>(candidate_receipt);
 
-    auto env_opt = makeCandidateEnvironment(
-        *session_keys_, *rolling_session_window_, session);
-    if (not env_opt.has_value()) {
-      SL_DEBUG(log_,
-               "We are lacking a `SessionInfo` for handling import of "
-               "statements.");
-      return outcome::success(false);
-    }
-    auto &env = env_opt.value();
-
     // In case we are not provided with a candidate receipt we operate under
     // the assumption, that a previous vote which included a
     // `CandidateReceipt` was seen. This holds since every block is preceded
@@ -1062,6 +1061,16 @@ namespace kagome::dispute {
       relay_parent = old_state_opt->candidate_receipt.descriptor.relay_parent;
     }
 
+    auto env_opt = makeCandidateEnvironment(
+        *session_keys_, *rolling_session_window_, session, relay_parent);
+    if (not env_opt.has_value()) {
+      SL_DEBUG(log_,
+               "We are lacking a `SessionInfo` for handling import of "
+               "statements.");
+      return outcome::success(false);
+    }
+    auto &env = env_opt.value();
+
     auto disabled_validators_res = api_->disabled_validators(relay_parent);
     if (disabled_validators_res.has_error()) {
       SL_WARN(log_,
@@ -1154,19 +1163,24 @@ namespace kagome::dispute {
                                    statements.end());
               }
               ++imported_valid_votes;
-              return true;
+              //              return true;
             }
-            auto &existing = std::get<0>(it->second);
-            return visit_in_place(
-                valid,
-                [&](const Explicit &) {
-                  return not is_type<Explicit>(existing);
-                },
-                [&](const BackingSeconded &) { return false; },
-                [&](const BackingValid &) { return false; },
-                [&](const ApprovalChecking &) {
-                  return not is_type<ApprovalChecking>(existing);
-                });
+            //            auto &existing = std::get<0>(it->second);
+            //            return visit_in_place(
+            //                valid,
+            //                [&](const Explicit &) {
+            //                  return not is_type<Explicit>(existing);
+            //                },
+            //                [&](const BackingSeconded &) { return false; },
+            //                [&](const BackingValid &) { return false; },
+            //                [&](const ApprovalChecking &) {
+            //                  return not is_type<ApprovalChecking>(existing);
+            //                },
+            //                [&](const ApprovalCheckingMultipleCandidates &) {
+            //                  return not
+            //                  is_type<ApprovalCheckingMultipleCandidates>(
+            //                      existing);
+            //                });
           },
           [&](const InvalidDisputeStatement &invalid) {
             auto [it, fresh] = votes.invalid.emplace(
@@ -1181,9 +1195,9 @@ namespace kagome::dispute {
                                    statements.end());
               }
               ++imported_invalid_votes;
-              return true;
+              //              return true;
             }
-            return false;
+            //            return false;
           });
     }
 
@@ -1467,7 +1481,11 @@ namespace kagome::dispute {
                 sig,
                 session};
 
-            SL_TRACE(log_, "Sending out own approval vote");
+            SL_TRACE(
+                log_,
+                "Sending out own approval vote. session={}, candidate_hash={}",
+                session,
+                candidate_hash);
 
             auto dispute_message_res = make_dispute_message(
                 env.session, new_state.votes, statement, validator_index);
@@ -1791,7 +1809,12 @@ namespace kagome::dispute {
       bool valid) {
     // https://github.com/paritytech/polkadot/blob/40974fb99c86f5c341105b7db53c7aa0df707d66/node/core/dispute-coordinator/src/initialized.rs#L1102
 
-    SL_TRACE(log_, "Issuing local statement for candidate!");
+    SL_TRACE(log_,
+             "Issuing local statement for candidate! "
+             "session={}, candidate_hash={}, relay_parent={}",
+             session,
+             candidate_hash,
+             candidate_receipt.descriptor.relay_parent);
 
     // Load environment:
 
@@ -2058,7 +2081,12 @@ namespace kagome::dispute {
              std::move(candidate_receipt),
              valid);
 
-    SL_TRACE(log_, "DisputeCoordinatorMessage::IssueLocalStatement");
+    SL_TRACE(log_,
+             "DisputeCoordinatorMessage::IssueLocalStatement. "
+             "session={}, candidate_hash={}, relay_parent={}",
+             session,
+             candidate_hash,
+             candidate_receipt.descriptor.relay_parent);
     auto res = issue_local_statement(
         candidate_hash, candidate_receipt, session, valid);
 
@@ -2507,6 +2535,12 @@ namespace kagome::dispute {
       }
     }
 
+    SL_TRACE(  // FIXME It's temporary code. Remove after tests
+        log_,
+        "Dispute (candidate={}) did not send. Disabled",
+        candidate_hash);
+    return;
+
     auto protocol = router_->getSendDisputeProtocol();
     BOOST_ASSERT_MSG(protocol,
                      "Router did not provide `send dispute` protocol");

core/dispute_coordinator/impl/dispute_coordinator_impl.hpp

@@ -192,10 +192,11 @@ namespace kagome::dispute {
         std::vector<Indexed<SignedDisputeStatement>> statements,
         CbOutcome<void> &&cb);
 
-    static std::optional<CandidateEnvironment> makeCandidateEnvironment(
+    std::optional<CandidateEnvironment> makeCandidateEnvironment(
         crypto::SessionKeys &session_keys,
         RollingSessionWindow &rolling_session_window,
-        SessionIndex session);
+        SessionIndex session,
+        primitives::BlockHash relay_parent);
 
     outcome::result<void> process_on_chain_votes(ScrapedOnChainVotes votes);