Feature: disabled validators in babe

core/consensus/babe/impl/babe.cpp

@@ -34,7 +34,6 @@
 #include "offchain/offchain_worker_factory.hpp"
 #include "offchain/offchain_worker_pool.hpp"
 #include "parachain/availability/bitfield/store.hpp"
-#include "parachain/backing/store.hpp"
 #include "parachain/parachain_inherent_data.hpp"
 #include "parachain/validator/parachain_processor.hpp"
 #include "primitives/inherent_data.hpp"
@@ -170,10 +169,26 @@ namespace kagome::consensus::babe {
     }
 
     const auto &authorities = config.value()->authorities;
-    if (session_keys_->getBabeKeyPair(authorities)) {
+    const auto &kp_opt = session_keys_->getBabeKeyPair(authorities);
+    if (kp_opt.has_value()) {
+      const auto &authority_index = kp_opt->second;
+      std::vector<AuthorityIndex> disabled_validators;
+      if (auto res = babe_api_->disabled_validators(block.hash);
+          res.has_value()) {
+        SL_CRITICAL(
+            log_, "Can't obtain disabled validators list for block {}", block);
+      }
+
       if (authorities.size() > 1) {
         return ValidatorStatus::Validator;
       }
+
+      if (std::binary_search(disabled_validators.begin(),
+                             disabled_validators.end(),
+                             authority_index)) {
+        return ValidatorStatus::DisabledValidator;
+      }
+
       return ValidatorStatus::SingleValidator;
     }
 
@@ -205,20 +220,23 @@ namespace kagome::consensus::babe {
     if (lottery_->getEpoch() != epoch) {
       is_active_validator_ = changeEpoch(epoch, best_block);
       metric_is_relaychain_validator_->set(is_active_validator_);
-      if (not is_active_validator_) {
-        if (is_validator_by_config_) {
-          SL_VERBOSE(log_,
-                     "Authority not known, skipping slot processing. "
-                     "Probably authority list has changed.");
-        }
+      if (not is_active_validator_ and is_validator_by_config_) {
+        SL_VERBOSE(log_,
+                   "Authority not known, skipping slot processing. "
+                   "Probably authority list has changed.");
       } else {
         SL_VERBOSE(log_, "Node is active validator in epoch {}", epoch);
       }
     }
 
     if (not is_active_validator_) {
       SL_TRACE(log_, "Node is not active validator in epoch {}", epoch);
-      return SlotLeadershipError::NO_VALIDATOR;
+      return SlotLeadershipError::NON_VALIDATOR;
+    }
+
+    auto validator_status = getValidatorStatus(best_block, epoch);
+    if (validator_status == ValidatorStatus::DisabledValidator) {
+      return SlotLeadershipError::DISABLED_VALIDATOR;
     }
 
     if (not checkSlotLeadership(best_block, slot)) {

core/consensus/babe/impl/babe_block_validator_impl.cpp

@@ -8,6 +8,7 @@
 
 #include <latch>
 
+#include "application/app_state_manager.hpp"
 #include "consensus/babe/babe_config_repository.hpp"
 #include "consensus/babe/babe_lottery.hpp"
 #include "consensus/babe/impl/babe_digests_util.hpp"
@@ -20,17 +21,19 @@
 #include "prepare_transcript.hpp"
 #include "primitives/inherent_data.hpp"
 #include "primitives/transcript.hpp"
+#include "runtime/runtime_api/babe_api.hpp"
 #include "runtime/runtime_api/offchain_worker_api.hpp"
-#include "storage/trie/serialization/ordered_trie_hash.hpp"
 #include "threshold_util.hpp"
 
 OUTCOME_CPP_DEFINE_CATEGORY(kagome::consensus::babe,
                             BabeBlockValidatorImpl::ValidationError,
                             e) {
   using E = kagome::consensus::babe::BabeBlockValidatorImpl::ValidationError;
   switch (e) {
-    case E::NO_AUTHORITIES:
-      return "no authorities are provided for the validation";
+    case E::NO_VALIDATOR:
+      return "author of block is not active validator";
+    case E::DISABLED_VALIDATOR:
+      return "author of block is disabled validator";
     case E::INVALID_SIGNATURE:
       return "SR25519 signature, which is in BABE header, is invalid";
     case E::INVALID_VRF:
@@ -46,21 +49,51 @@ OUTCOME_CPP_DEFINE_CATEGORY(kagome::consensus::babe,
 namespace kagome::consensus::babe {
 
   BabeBlockValidatorImpl::BabeBlockValidatorImpl(
+      std::shared_ptr<application::AppStateManager> app_state_manager,
       LazySPtr<SlotsUtil> slots_util,
       std::shared_ptr<BabeConfigRepository> config_repo,
       std::shared_ptr<crypto::Hasher> hasher,
       std::shared_ptr<crypto::Sr25519Provider> sr25519_provider,
-      std::shared_ptr<crypto::VRFProvider> vrf_provider)
+      std::shared_ptr<crypto::VRFProvider> vrf_provider,
+      std::shared_ptr<runtime::BabeApi> babe_api,
+      primitives::events::SyncStateSubscriptionEnginePtr sync_state_observable)
       : log_(log::createLogger("BabeBlockValidatorImpl", "babe")),
         slots_util_(std::move(slots_util)),
         config_repo_(std::move(config_repo)),
         hasher_(std::move(hasher)),
         sr25519_provider_(std::move(sr25519_provider)),
-        vrf_provider_(std::move(vrf_provider)) {
+        vrf_provider_(std::move(vrf_provider)),
+        babe_api_(std::move(babe_api)),
+        sync_state_observable_(std::move(sync_state_observable)) {
     BOOST_ASSERT(config_repo_);
     BOOST_ASSERT(hasher_);
     BOOST_ASSERT(sr25519_provider_);
     BOOST_ASSERT(vrf_provider_);
+    BOOST_ASSERT(babe_api_);
+    BOOST_ASSERT(sync_state_observable_);
+
+    app_state_manager->takeControl(*this);
+  }
+
+  void BabeBlockValidatorImpl::prepare() {
+    sync_state_observer_ =
+        std::make_shared<primitives::events::SyncStateEventSubscriber>(
+            sync_state_observable_, false);
+    sync_state_observer_->subscribe(
+        sync_state_observer_->generateSubscriptionSetId(),
+        primitives::events::SyncStateEventType::kSyncState);
+    sync_state_observer_->setCallback(
+        [wp{weak_from_this()}](
+            auto /*set_id*/,
+            bool &synchronized,
+            auto /*event_type*/,
+            const primitives::events::SyncStateEventParams &event) mutable {
+          if (auto self = wp.lock()) {
+            if (event == consensus::SyncState::SYNCHRONIZED) {
+              self->was_synchronized_ = true;
+            }
+          }
+        });
   }
 
   outcome::result<void> BabeBlockValidatorImpl::validateHeader(
@@ -94,6 +127,10 @@ namespace kagome::consensus::babe {
              epoch_number,
              config.randomness);
 
+    if (babe_header.authority_index >= config.authorities.size()) {
+      return ValidationError::NO_VALIDATOR;
+    }
+
     auto threshold = calculateThreshold(config.leadership_rate,
                                         config.authorities,
                                         babe_header.authority_index);
@@ -105,6 +142,24 @@ namespace kagome::consensus::babe {
                        threshold,
                        config));
 
+    // If we were synchronized,
+    // we have available runtime to check disabled validators
+    if (was_synchronized_) {
+      std::vector<AuthorityIndex> disabled_validators;
+      if (auto res = babe_api_->disabled_validators(block_header.parent_hash);
+          res.has_value()) {
+        SL_CRITICAL(log_,
+                    "Can't obtain disabled validators list for block {}",
+                    block_header.blockInfo());
+      }
+
+      if (std::binary_search(disabled_validators.begin(),
+                             disabled_validators.end(),
+                             babe_header.authority_index)) {
+        return ValidationError::DISABLED_VALIDATOR;
+      }
+    }
+
     return outcome::success();
   }
 

core/consensus/babe/impl/babe_block_validator_impl.hpp

@@ -18,6 +18,10 @@
 #include "primitives/event_types.hpp"
 #include "telemetry/service.hpp"
 
+namespace kagome::application {
+  class AppStateManager;
+}
+
 namespace kagome::consensus {
   class SlotsUtil;
 }
@@ -33,24 +37,35 @@ namespace kagome::crypto {
   class VRFProvider;
 }  // namespace kagome::crypto
 
+namespace kagome::runtime {
+  class BabeApi;
+}
+
 namespace kagome::consensus::babe {
 
   class BabeBlockValidatorImpl
       : public BabeBlockValidator,
         public std::enable_shared_from_this<BabeBlockValidatorImpl> {
    public:
     BabeBlockValidatorImpl(
+        std::shared_ptr<application::AppStateManager> app_state_manager,
         LazySPtr<SlotsUtil> slots_util,
         std::shared_ptr<BabeConfigRepository> config_repo,
         std::shared_ptr<crypto::Hasher> hasher,
         std::shared_ptr<crypto::Sr25519Provider> sr25519_provider,
-        std::shared_ptr<crypto::VRFProvider> vrf_provider);
+        std::shared_ptr<crypto::VRFProvider> vrf_provider,
+        std::shared_ptr<runtime::BabeApi> babe_api,
+        primitives::events::SyncStateSubscriptionEnginePtr
+            sync_state_observable);
+
+    void prepare();
 
     outcome::result<void> validateHeader(
         const primitives::BlockHeader &block_header) const;
 
     enum class ValidationError {
-      NO_AUTHORITIES = 1,
+      NO_VALIDATOR = 1,
+      DISABLED_VALIDATOR,
       INVALID_SIGNATURE,
       INVALID_VRF,
       TWO_BLOCKS_IN_SLOT,
@@ -107,6 +122,11 @@ namespace kagome::consensus::babe {
     std::shared_ptr<crypto::Hasher> hasher_;
     std::shared_ptr<crypto::Sr25519Provider> sr25519_provider_;
     std::shared_ptr<crypto::VRFProvider> vrf_provider_;
+    std::shared_ptr<runtime::BabeApi> babe_api_;
+    primitives::events::SyncStateSubscriptionEnginePtr sync_state_observable_;
+
+    bool was_synchronized_ = false;
+    primitives::events::SyncStateEventSubscriberPtr sync_state_observer_;
   };
 
 }  // namespace kagome::consensus::babe

core/consensus/babe/impl/threshold_util.cpp

@@ -14,6 +14,8 @@ namespace kagome::consensus::babe {
   Threshold calculateThreshold(const std::pair<uint64_t, uint64_t> &ratio,
                                const Authorities &authorities,
                                AuthorityIndex authority_index) {
+    BOOST_ASSERT(authority_index < authorities.size());
+
     double float_point_ratio = double(ratio.first) / ratio.second;
 
     using boost::adaptors::transformed;

core/consensus/babe/types/authority.hpp

@@ -24,6 +24,4 @@ namespace kagome::consensus::babe {
 
   using Authorities = common::SLVector<Authority, kMaxValidatorsNumber>;
 
-  using AuthorityIndex = uint32_t;
-
 }  // namespace kagome::consensus::babe

core/consensus/grandpa/types/authority.hpp

@@ -25,8 +25,6 @@ namespace kagome::consensus::grandpa {
   using Authorities =
       common::SLVector<Authority, consensus::kMaxValidatorsNumber>;
 
-  using AuthorityIndex = uint32_t;
-
   using AuthoritySetId = uint64_t;
 
   /*

core/consensus/grandpa/voter_set.hpp

@@ -11,6 +11,7 @@
 #include "common/outcome_throw.hpp"
 #include "consensus/grandpa/common.hpp"
 #include "consensus/grandpa/types/authority.hpp"
+#include "consensus/timeline/types.hpp"
 
 namespace kagome::consensus::grandpa {
 

core/consensus/production_consensus.hpp

@@ -14,7 +14,7 @@ namespace kagome::consensus {
 
   enum class ValidatorStatus {
     NonValidator = 0,
-    InactiveValidator,
+    DisabledValidator,
     Validator,
     SingleValidator,
   };

core/consensus/timeline/impl/block_appender_base.cpp

@@ -75,9 +75,9 @@ namespace kagome::consensus {
 
   outcome::result<BlockAppenderBase::SlotInfo> BlockAppenderBase::getSlotInfo(
       const primitives::BlockHeader &header) const {
-    OUTCOME_TRY(
-        slot_number,
-        babe::getSlot(header));  // TODO(xDimon): Make it consensus agnostic
+    auto consensus = consensus_selector_.get()->getProductionConsensus(header);
+    BOOST_ASSERT_MSG(consensus, "Must be returned at least fallback consensus");
+    OUTCOME_TRY(slot_number, consensus->getSlot(header));
     auto start_time = slots_util_.get()->slotStartTime(slot_number);
     auto slot_duration = timings_.slot_duration;
     return outcome::success(SlotInfo{start_time, slot_duration});

core/consensus/timeline/impl/slot_leadership_error.cpp

@@ -9,8 +9,10 @@
 OUTCOME_CPP_DEFINE_CATEGORY(kagome::consensus, SlotLeadershipError, e) {
   using E = kagome::consensus::SlotLeadershipError;
   switch (e) {
-    case E::NO_VALIDATOR:
+    case E::NON_VALIDATOR:
       return "node is not validator in current epoch";
+    case E::DISABLED_VALIDATOR:
+      return "node is disabled validator till end of epoch";
     case E::NO_SLOT_LEADER:
       return "node is not slot leader in current slot";
     case E::BACKING_OFF:

core/consensus/timeline/impl/slot_leadership_error.hpp

@@ -11,7 +11,8 @@
 namespace kagome::consensus {
 
   enum class SlotLeadershipError {
-    NO_VALIDATOR = 1,
+    NON_VALIDATOR = 1,
+    DISABLED_VALIDATOR,
     NO_SLOT_LEADER,
     BACKING_OFF,
   };

core/consensus/timeline/impl/timeline_impl.cpp

@@ -52,7 +52,7 @@ namespace kagome::consensus {
           justification_observer,
       std::shared_ptr<libp2p::basic::Scheduler> scheduler,
       primitives::events::ChainSubscriptionEnginePtr chain_sub_engine,
-      primitives::events::BabeStateSubscriptionEnginePtr state_sub_engine,
+      primitives::events::SyncStateSubscriptionEnginePtr state_sub_engine,
       std::shared_ptr<runtime::Core> core_api)
       : log_(log::createLogger("Timeline", "timeline")),
         app_state_manager_(std::move(app_state_manager)),
@@ -361,7 +361,7 @@ namespace kagome::consensus {
       return;
     }
 
-    // Start catching up if gap recognized
+    // Start catching up if a gap recognized
     if (current_state_ == SyncState::SYNCHRONIZED
         or current_state_ == SyncState::HEADERS_LOADED) {
       if (announce.header.number > current_best_block.number + 1) {
@@ -371,7 +371,7 @@ namespace kagome::consensus {
     }
 
     // Received announce that has the same block number as ours best,
-    // or greater by one. Using of simple way to load block
+    // or greater by one. Using a simple way to load block
     synchronizer_->syncByBlockHeader(
         announce.header,
         peer_id,

core/consensus/timeline/impl/timeline_impl.hpp

@@ -86,7 +86,7 @@ namespace kagome::consensus {
             justification_observer,
         std::shared_ptr<libp2p::basic::Scheduler> scheduler,
         primitives::events::ChainSubscriptionEnginePtr chain_sub_engine,
-        primitives::events::BabeStateSubscriptionEnginePtr state_sub_engine,
+        primitives::events::SyncStateSubscriptionEnginePtr state_sub_engine,
         std::shared_ptr<runtime::Core> core_api);
 
     /// @see AppStateManager::takeControl
@@ -154,7 +154,7 @@ namespace kagome::consensus {
     std::shared_ptr<libp2p::basic::Scheduler> scheduler_;
     primitives::events::ChainSubscriptionEnginePtr chain_sub_engine_;
     primitives::events::ChainSub chain_sub_;
-    primitives::events::BabeStateSubscriptionEnginePtr state_sub_engine_;
+    primitives::events::SyncStateSubscriptionEnginePtr state_sub_engine_;
     std::shared_ptr<runtime::Core> core_api_;
 
     application::SyncMethod sync_method_;

core/consensus/timeline/types.hpp

@@ -25,6 +25,8 @@ namespace kagome::consensus {
   /// slot number of the block production
   using SlotNumber = uint64_t;
 
+  using AuthorityIndex = uint32_t;
+
   /// duration of single slot in milliseconds
   struct SlotDuration : public std::chrono::milliseconds {
     SlotDuration() = default;

core/dispute_coordinator/dispute_coordinator.hpp

@@ -40,12 +40,6 @@ namespace kagome::dispute {
 
     virtual ~DisputeCoordinator() = default;
 
-    //    virtual void importStatements(CandidateReceipt candidate_receipt,
-    //                                  SessionIndex session,
-    //                                  std::vector<Indexed<SignedDisputeStatement>>
-    //                                      statements,  // FIXME avoid copy
-    //                                  CbOutcome<void> &&cb) = 0;
-
     /// Fetch a list of all recent disputes the coordinator is aware of.
     /// These are disputes which have occurred any time in recent sessions,
     /// and which may have already concluded.