feat(consensus): [CON-1173] Purge non-finalized blocks and notarizati…

…ons below the finalized height

rs/artifact_pool/src/consensus_pool.rs

@@ -31,10 +31,15 @@ use std::{marker::PhantomData, sync::Arc, time::Duration};
 
 #[derive(Debug, Clone)]
 pub enum PoolSectionOp<T> {
+    /// Insert the artifact into the pool section.
     Insert(T),
+    /// Remove the artifact with the given [`ConsensusMessageId`] from the pool section.
     Remove(ConsensusMessageId),
-    PurgeBelow(Height),                            // Non-inclusive
-    PurgeTypeBelow(PurgeableArtifactType, Height), // Non-inclusive
+    /// Remove all the artifacts _strictly_ below the height from the pool section.
+    PurgeBelow(Height),
+    /// Remove all the artifacts of the given type _strictly_ below the height from the pool
+    /// section.
+    PurgeTypeBelow(PurgeableArtifactType, Height),
 }
 
 #[derive(Clone, Debug, Default)]
@@ -46,16 +51,29 @@ impl<T> PoolSectionOps<T> {
     pub fn new() -> PoolSectionOps<T> {
         PoolSectionOps { ops: Vec::new() }
     }
+
+    /// Insert the artifact into the pool section.
     pub fn insert(&mut self, artifact: T) {
         self.ops.push(PoolSectionOp::Insert(artifact));
     }
-    pub fn remove(&mut self, msg_id: ConsensusMessageId) {
+
+    /// Remove the artifact with the given [`ConsensusMessageId`] from the pool section.
+    pub(crate) fn remove(&mut self, msg_id: ConsensusMessageId) {
         self.ops.push(PoolSectionOp::Remove(msg_id));
     }
-    pub fn purge_below(&mut self, height: Height) {
+
+    /// Remove all the artifacts _strictly_ below the height from the pool section.
+    pub(crate) fn purge_below(&mut self, height: Height) {
         self.ops.push(PoolSectionOp::PurgeBelow(height));
     }
-    pub fn purge_type_below(&mut self, artifact_type: PurgeableArtifactType, height: Height) {
+
+    /// Remove all the artifacts of the given type _strictly_ below the height from the pool
+    /// section.
+    pub(crate) fn purge_type_below(
+        &mut self,
+        artifact_type: PurgeableArtifactType,
+        height: Height,
+    ) {
         self.ops
             .push(PoolSectionOp::PurgeTypeBelow(artifact_type, height));
     }
@@ -69,6 +87,7 @@ pub trait MutablePoolSection<T>: PoolSection<T> {
     /// Mutate the pool by applying the given [`PoolSectionOps`]. Return [`ConsensusMessageId`]s
     /// of artifacts that were deleted during the mutation.
     fn mutate(&mut self, ops: PoolSectionOps<T>) -> Vec<ConsensusMessageId>;
+
     /// Return a reference to the [`PoolSection`].
     fn pool_section(&self) -> &dyn PoolSection<T>;
 }
@@ -610,6 +629,9 @@ impl MutablePool<ConsensusArtifact> for ConsensusPoolImpl {
                     adverts.push(ConsensusArtifact::message_to_advert(&to_add.msg));
                     validated_ops.insert(to_add);
                 }
+                ChangeAction::RemoveFromValidated(to_remove) => {
+                    validated_ops.remove(to_remove.get_id());
+                }
                 ChangeAction::MoveToValidated(to_move) => {
                     if !to_move.is_share() {
                         adverts.push(ConsensusArtifact::message_to_advert(&to_move));

rs/artifact_pool/src/consensus_pool_cache.rs

@@ -13,7 +13,7 @@ use std::cmp::Ordering;
 use std::collections::BTreeMap;
 use std::sync::{Arc, RwLock};
 
-/// Implementation of ConsensusCache and ConsensusPoolCache.
+/// Implementation of [`ConsensusBlockCache`] and [`ConsensusPoolCache`].
 pub(crate) struct ConsensusCacheImpl {
     cache: RwLock<CachedData>,
 }
@@ -25,7 +25,7 @@ pub(crate) enum CacheUpdateAction {
     CatchUpPackage,
 }
 
-// Internal cached data held by the the ConsensusCache.
+/// Internal cached data held by the [`ConsensusCacheImpl`].
 struct CachedData {
     finalized_block: Block,
     summary_block: Block,

rs/consensus/src/consensus/block_maker.rs

@@ -24,9 +24,8 @@ use ic_types::{
     batch::{BatchPayload, ValidationContext},
     consensus::{
         block_maker::SubnetRecords, dkg, hashed, Block, BlockPayload, BlockProposal, DataPayload,
-        HasRank, Payload, RandomBeacon, Rank, SummaryPayload,
+        HasHeight, HasRank, HashedBlock, Payload, RandomBeacon, Rank, SummaryPayload,
     },
-    crypto::CryptoHashOf,
     replica_config::ReplicaConfig,
     time::current_time,
     CountBytes, Height, NodeId, RegistryVersion,
@@ -184,10 +183,9 @@ impl BlockMaker {
         &self,
         pool: &PoolReader<'_>,
         rank: Rank,
-        parent: Block,
+        parent: HashedBlock,
     ) -> Option<BlockProposal> {
-        let parent_hash = ic_types::crypto::crypto_hash(&parent);
-        let height = parent.height.increment();
+        let height = parent.height().increment();
         let certified_height = self.state_manager.latest_certified_height();
 
         // Note that we will skip blockmaking if registry versions or replica_versions
@@ -207,7 +205,7 @@ impl BlockMaker {
 
         // The stable registry version to be agreed on in this block. If this is a summary
         // block, this version will be the new membership version of the next dkg interval.
-        let stable_registry_version = self.get_stable_registry_version(&parent)?;
+        let stable_registry_version = self.get_stable_registry_version(parent.as_ref())?;
         // Get the subnet records that are relevant to making a block
         let subnet_records =
             subnet_records_for_registry_version(self, registry_version, stable_registry_version)?;
@@ -251,11 +249,11 @@ impl BlockMaker {
             // blocks. The additional 1ns makes no practical difference in that regard.
             time: std::cmp::max(
                 self.time_source.get_relative_time(),
-                parent.context.time + monotonic_block_increment,
+                parent.as_ref().context.time + monotonic_block_increment,
             ),
         };
 
-        if !context.greater(&parent.context) {
+        if !context.greater(&parent.as_ref().context) {
             // The values in our validation context are not strictly monotonically
             // increasing the values included in the parent block by at least
             // monotonic_block_increment. To avoid proposing an invalid block, we simply
@@ -267,7 +265,7 @@ impl BlockMaker {
                 smaller than the parent validation context (locally available={:?}, \
                 parent context={:?})",
                 context,
-                &parent.context
+                &parent.as_ref().context
             );
             return None;
         }
@@ -276,7 +274,6 @@ impl BlockMaker {
             pool,
             context,
             parent,
-            parent_hash,
             height,
             certified_height,
             rank,
@@ -293,8 +290,7 @@ impl BlockMaker {
         &self,
         pool: &PoolReader<'_>,
         context: ValidationContext,
-        parent: Block,
-        parent_hash: CryptoHashOf<Block>,
+        parent: HashedBlock,
         height: Height,
         certified_height: Height,
         rank: Rank,
@@ -310,7 +306,7 @@ impl BlockMaker {
             &*self.crypto,
             pool,
             Arc::clone(&self.dkg_pool),
-            &parent,
+            parent.as_ref(),
             &*self.state_manager,
             &context,
             self.log.clone(),
@@ -330,7 +326,7 @@ impl BlockMaker {
                         &*self.registry_client,
                         pool,
                         &context,
-                        &parent,
+                        parent.as_ref(),
                         Some(&self.ecdsa_payload_metrics),
                         &self.log,
                     )
@@ -367,7 +363,7 @@ impl BlockMaker {
                                 height,
                                 certified_height,
                                 &context,
-                                &parent,
+                                parent.as_ref(),
                                 subnet_records,
                             );
 
@@ -379,7 +375,7 @@ impl BlockMaker {
                                 self.ecdsa_pool.clone(),
                                 &*self.state_manager,
                                 &context,
-                                &parent,
+                                parent.as_ref(),
                                 &self.ecdsa_payload_metrics,
                                 &self.log,
                             )
@@ -406,7 +402,7 @@ impl BlockMaker {
                 }
             },
         );
-        let block = Block::new(parent_hash, payload, height, rank, context);
+        let block = Block::new(parent.get_hash().clone(), payload, height, rank, context);
         let hashed_block = hashed::Hashed::new(ic_types::crypto::crypto_hash, block);
         match self
             .crypto
@@ -498,7 +494,7 @@ impl BlockMaker {
 /// Return the parent random beacon and block of the latest round for which
 /// this node might propose a block.
 /// Return None otherwise.
-pub(crate) fn get_dependencies(pool: &PoolReader<'_>) -> Option<(RandomBeacon, Block)> {
+pub(crate) fn get_dependencies(pool: &PoolReader<'_>) -> Option<(RandomBeacon, HashedBlock)> {
     let notarized_height = pool.get_notarized_height();
     let beacon = pool.get_random_beacon(notarized_height)?;
     let parent = pool

rs/consensus/src/consensus/finalizer.rs

@@ -31,7 +31,7 @@ use ic_interfaces_registry::RegistryClient;
 use ic_logger::{debug, trace, ReplicaLogger};
 use ic_metrics::MetricsRegistry;
 use ic_types::{
-    consensus::{Block, FinalizationContent, FinalizationShare},
+    consensus::{FinalizationContent, FinalizationShare, HashedBlock},
     replica_config::ReplicaConfig,
     Height, ReplicaVersion,
 };
@@ -156,7 +156,7 @@ impl Finalizer {
     ///
     /// In this case, the single notarized block is returned. Otherwise,
     /// return `None`
-    fn pick_block_to_finality_sign(&self, pool: &PoolReader<'_>, h: Height) -> Option<Block> {
+    fn pick_block_to_finality_sign(&self, pool: &PoolReader<'_>, h: Height) -> Option<HashedBlock> {
         let me = self.replica_config.node_id;
         let previous_beacon = pool.get_random_beacon(h.decrement())?;
         // check whether this replica was a notary at height h
@@ -200,10 +200,9 @@ impl Finalizer {
 
         // If notarization shares exists created by this replica at height `h`
         // that sign a block different than `notarized_block`, do not finalize.
-        let other_notarized_shares_exists = pool.get_notarization_shares(h).any(|x| {
-            x.signature.signer == me
-                && x.content.block != ic_types::crypto::crypto_hash(&notarized_block)
-        });
+        let other_notarized_shares_exists = pool
+            .get_notarization_shares(h)
+            .any(|x| x.signature.signer == me && x.content.block != *notarized_block.get_hash());
         if other_notarized_shares_exists {
             return None;
         }
@@ -216,7 +215,9 @@ impl Finalizer {
     fn finalize_height(&self, pool: &PoolReader<'_>, height: Height) -> Option<FinalizationShare> {
         let content = FinalizationContent::new(
             height,
-            ic_types::crypto::crypto_hash(&self.pick_block_to_finality_sign(pool, height)?),
+            self.pick_block_to_finality_sign(pool, height)?
+                .get_hash()
+                .clone(),
         );
         let signature = self
             .crypto

rs/consensus/src/consensus/malicious_consensus.rs

@@ -10,7 +10,7 @@ use ic_interfaces::consensus_pool::{ChangeAction, ChangeSet, HeightRange};
 use ic_logger::{info, trace, ReplicaLogger};
 use ic_types::consensus::{
     hashed, Block, BlockProposal, ConsensusMessage, ConsensusMessageHashable, FinalizationContent,
-    FinalizationShare, NotarizationShare, Rank,
+    FinalizationShare, HasHeight, HashedBlock, NotarizationShare, Rank,
 };
 use ic_types::malicious_flags::MaliciousFlags;
 use ic_types::Time;
@@ -156,19 +156,18 @@ fn maliciously_propose_empty_block(
     block_maker: &BlockMaker,
     pool: &PoolReader<'_>,
     rank: Rank,
-    parent: Block,
+    parent: HashedBlock,
 ) -> Option<BlockProposal> {
-    let parent_hash = ic_types::crypto::crypto_hash(&parent);
-    let height = parent.height.increment();
+    let height = parent.height().increment();
     let certified_height = block_maker.state_manager.latest_certified_height();
-    let context = parent.context.clone();
+    let context = parent.as_ref().context.clone();
 
     // Note that we will skip blockmaking if registry versions or replica_versions
     // are missing or temporarily not retrievable.
     let registry_version = pool.registry_version(height)?;
 
     // Get the subnet records that are relevant to making a block
-    let stable_registry_version = block_maker.get_stable_registry_version(&parent)?;
+    let stable_registry_version = block_maker.get_stable_registry_version(parent.as_ref())?;
     let subnet_records = block_maker::subnet_records_for_registry_version(
         block_maker,
         registry_version,
@@ -179,7 +178,6 @@ fn maliciously_propose_empty_block(
         pool,
         context,
         parent,
-        parent_hash,
         height,
         certified_height,
         rank,