historical votes

core/consensus/grandpa/grandpa.hpp

@@ -8,6 +8,7 @@
 
 #include "consensus/finality_consensus.hpp"
 #include "consensus/grandpa/common.hpp"
+#include "consensus/grandpa/historical_votes.hpp"
 
 #include <memory>
 
@@ -21,7 +22,7 @@ namespace kagome::consensus::grandpa {
    * Interface for launching new grandpa rounds. See more details in
    * kagome::consensus::grandpa::GrandpaImpl
    */
-  class Grandpa : public FinalityConsensus {
+  class Grandpa : public FinalityConsensus, public SaveHistoricalVotes {
    public:
     virtual ~Grandpa() = default;
 

core/consensus/grandpa/historical_votes.hpp

@@ -0,0 +1,35 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include "consensus/grandpa/structs.hpp"
+
+namespace kagome::consensus::grandpa {
+  /**
+   * Historical votes seen in a round.
+   * https://github.com/paritytech/finality-grandpa/blob/8c45a664c05657f0c71057158d3ba555ba7d20de/src/lib.rs#L544
+   */
+  struct HistoricalVotes {
+    SCALE_TIE(3);
+
+    std::vector<SignedMessage> seen;
+    std::optional<uint64_t> prevote_idx, precommit_idx;
+  };
+
+  class SaveHistoricalVotes {
+   public:
+    virtual ~SaveHistoricalVotes() = default;
+
+    /**
+     * Called from `VotingRoundImpl` to `GrandpaImpl` to save historical vote.
+     */
+    virtual void saveHistoricalVote(AuthoritySetId set,
+                                    RoundNumber round,
+                                    const SignedMessage &vote,
+                                    bool set_index) = 0;
+  };
+}  // namespace kagome::consensus::grandpa

core/consensus/grandpa/impl/grandpa_impl.cpp

@@ -66,6 +66,13 @@ namespace kagome::consensus::grandpa {
   // https://github.com/paritytech/polkadot/pull/6217
   constexpr std::chrono::milliseconds kGossipDuration{1000};
 
+  inline auto historicalVotesKey(AuthoritySetId set, RoundNumber round) {
+    auto key = storage::kGrandpaHistoricalVotesPrefix;
+    key.putUint64(set);
+    key.putUint64(round);
+    return key;
+  }
+
   GrandpaImpl::GrandpaImpl(
       std::shared_ptr<application::AppStateManager> app_state_manager,
       std::shared_ptr<crypto::Hasher> hasher,
@@ -125,12 +132,6 @@ namespace kagome::consensus::grandpa {
   }
 
   bool GrandpaImpl::tryStart() {
-    if (auto r = db_->get(storage::kGrandpaVotesKey)) {
-      if (auto r2 = scale::decode<CachedVotes>(r.value())) {
-        cached_votes_ = std::move(r2.value());
-      }
-    }
-
     // Obtain last completed round
     auto round_state_res = getLastCompletedRound();
     if (not round_state_res.has_value()) {
@@ -211,24 +212,18 @@ namespace kagome::consensus::grandpa {
     auto vote_crypto_provider = std::make_shared<VoteCryptoProviderImpl>(
         keypair, crypto_provider_, round_state.round_number, config.voters);
 
-    auto save_cached_votes = [weak{weak_from_this()}]() {
-      if (auto self = weak.lock()) {
-        self->saveCachedVotes();
-      }
-    };
     auto new_round = std::make_shared<VotingRoundImpl>(
         shared_from_this(),
         std::move(config),
         hasher_,
         environment_,
-        std::move(save_cached_votes),
         std::move(vote_crypto_provider),
         std::make_shared<VoteTrackerImpl>(),  // Prevote tracker
         std::make_shared<VoteTrackerImpl>(),  // Precommit tracker
         std::move(vote_graph),
         scheduler_,
         round_state);
-    applyCachedVotes(*new_round);
+    applyHistoricalVotes(*new_round);
 
     new_round->end();  // it is okay, because we do not want to actually execute
                        // this round
@@ -277,24 +272,18 @@ namespace kagome::consensus::grandpa {
     auto vote_crypto_provider = std::make_shared<VoteCryptoProviderImpl>(
         keypair, crypto_provider_, new_round_number, config.voters);
 
-    auto save_cached_votes = [weak{weak_from_this()}]() {
-      if (auto self = weak.lock()) {
-        self->saveCachedVotes();
-      }
-    };
     auto new_round = std::make_shared<VotingRoundImpl>(
         shared_from_this(),
         std::move(config),
         hasher_,
         environment_,
-        std::move(save_cached_votes),
         std::move(vote_crypto_provider),
         std::make_shared<VoteTrackerImpl>(),  // Prevote tracker
         std::make_shared<VoteTrackerImpl>(),  // Precommit tracker
         std::move(vote_graph),
         scheduler_,
         round);
-    applyCachedVotes(*new_round);
+    applyHistoricalVotes(*new_round);
     return new_round;
   }
 
@@ -1192,7 +1181,6 @@ namespace kagome::consensus::grandpa {
         GrandpaConfig{voters, justification.round_number, {}, {}},
         hasher_,
         environment_,
-        nullptr,
         std::make_shared<VoteCryptoProviderImpl>(
             nullptr, crypto_provider_, justification.round_number, voters),
         std::make_shared<VoteTrackerImpl>(),
@@ -1424,33 +1412,82 @@ namespace kagome::consensus::grandpa {
     retain_if(waiting_blocks_, f);
   }
 
-  void GrandpaImpl::saveCachedVotes() {
-    CachedVotes rounds;
-    for (auto round = current_round_; round;
-         round = round->getPreviousRound()) {
-      rounds.emplace_back(CachedRound{
-          round->voterSetId(),
-          round->roundNumber(),
-          round->votes(),
-      });
-    }
-    std::ignore =
-        db_->put(storage::kGrandpaVotesKey, scale::encode(rounds).value());
+  void GrandpaImpl::saveHistoricalVote(AuthoritySetId set,
+                                       RoundNumber round,
+                                       const SignedMessage &vote,
+                                       bool set_index) {
+    REINVOKE(*grandpa_pool_handler_,
+             saveHistoricalVote,
+             set,
+             round,
+             vote,
+             set_index);
+    auto &[votes, dirty] = historicalVotes(set, round);
+    if (std::find(votes.seen.begin(), votes.seen.end(), vote)
+        != votes.seen.end()) {
+      return;
+    }
+    if (set_index) {
+      auto *index = vote.is<Prevote>()   ? &votes.prevote_idx
+                  : vote.is<Precommit>() ? &votes.precommit_idx
+                                         : nullptr;
+      if (index and not *index) {
+        *index = votes.seen.size();
+      }
+    }
+    votes.seen.emplace_back(vote);
+    dirty = true;
+    if (writing_historical_votes_) {
+      return;
+    }
+    writing_historical_votes_ = true;
+    grandpa_pool_handler_->execute([weak_self{weak_from_this()}] {
+      auto self = weak_self.lock();
+      if (not self) {
+        return;
+      }
+      self->writeHistoricalVotes();
+    });
   }
 
-  void GrandpaImpl::applyCachedVotes(VotingRound &round) {
-    auto it = std::find_if(
-        cached_votes_.begin(), cached_votes_.end(), [&](const CachedRound &c) {
-          return c.set == round.voterSetId() and c.round == round.roundNumber();
+  void GrandpaImpl::writeHistoricalVotes() {
+    writing_historical_votes_ = false;
+    historical_votes_.forEach(
+        [&](const HistoricalVotesKey &key, HistoricalVotesDirty &cache) {
+          if (not cache.second) {
+            return;
+          }
+          cache.second = false;
+          std::ignore = db_->put(historicalVotesKey(key.first, key.second),
+                                 scale::encode(cache.first).value());
         });
-    if (it == cached_votes_.end()) {
-      return;
+  }
+
+  GrandpaImpl::HistoricalVotesDirty &GrandpaImpl::historicalVotes(
+      AuthoritySetId set, RoundNumber round) {
+    auto key = std::make_pair(set, round);
+    auto cache = historical_votes_.get(key);
+    if (not cache) {
+      cache = historical_votes_.put(key, {{}, false});
+      if (auto r = db_->get(historicalVotesKey(set, round))) {
+        if (auto r2 = scale::decode<HistoricalVotes>(r.value())) {
+          cache->get().first = std::move(r2.value());
+        } else {
+          SL_ERROR(logger_,
+                   "historicalVotes(set={}, round={}): decode error",
+                   set,
+                   round);
+        }
+      }
     }
+    return cache->get();
+  }
+
+  void GrandpaImpl::applyHistoricalVotes(VotingRound &round) {
+    auto &votes =
+        historicalVotes(round.voterSetId(), round.roundNumber()).first;
     VotingRoundUpdate update{round};
-    for (auto &vote : it->votes.first) {
-      update.vote(vote);
-    }
-    for (auto &vote : it->votes.second) {
+    for (auto &vote : votes.seen) {
       update.vote(vote);
     }
     update.update();