fix(consensus): create CUP shares without waiting for finalized tip to point to CUP height

rs/artifact_pool/src/consensus_pool_cache.rs

@@ -7,7 +7,7 @@ use ic_interfaces::consensus_pool::{
 use ic_protobuf::types::v1 as pb;
 use ic_types::{
     Height, Time,
-    consensus::{Block, CatchUpPackage, ConsensusMessage, Finalization, HasHeight, HashedBlock},
+    consensus::{Block, CatchUpPackage, ConsensusMessage, Finalization, HasHeight},
 };
 use std::cmp::Ordering;
 use std::collections::BTreeMap;
@@ -59,7 +59,6 @@ impl CachedData {
 struct CachedChainIterator<'a> {
     consensus_pool: &'a dyn ConsensusPool,
     finalized_chain: Arc<dyn ConsensusBlockChain>,
-    to_block: Option<HashedBlock>,
     cursor: Option<Block>,
 }
 
@@ -68,12 +67,10 @@ impl<'a> CachedChainIterator<'a> {
         consensus_pool: &'a dyn ConsensusPool,
         finalized_chain: Arc<dyn ConsensusBlockChain>,
         from_block: Block,
-        to_block: Option<HashedBlock>,
     ) -> Self {
         CachedChainIterator {
             consensus_pool,
             finalized_chain,
-            to_block,
             cursor: Some(from_block),
         }
     }
@@ -85,21 +82,6 @@ impl<'a> CachedChainIterator<'a> {
         }
         let parent_height = height.decrement();
         let parent_hash = &block.parent;
-        if let Some(to_block) = &self.to_block {
-            match parent_height.cmp(&to_block.height()) {
-                std::cmp::Ordering::Less => {
-                    return None;
-                }
-                std::cmp::Ordering::Equal => {
-                    if parent_hash == to_block.get_hash() {
-                        return Some(to_block.as_ref().clone());
-                    } else {
-                        return None;
-                    }
-                }
-                _ => (),
-            }
-        }
         // Use cached blocks if the height is finalized
         if parent_height <= self.finalized_chain.tip().height()
             && let Ok(block) = self.finalized_chain.get_block_by_height(parent_height)
@@ -183,7 +165,6 @@ impl ConsensusPoolCache for ConsensusCacheImpl {
             pool,
             self.finalized_chain(),
             block,
-            Some(self.catch_up_package().content.block),
         ))
     }
 }
@@ -335,7 +316,7 @@ pub(crate) fn update_summary_block(
             }
 
             // Otherwise, find the parent block at start_height
-            *summary_block = ChainIterator::new(consensus_pool, finalized_tip.clone(), None)
+            *summary_block = ChainIterator::new(consensus_pool, finalized_tip.clone())
                 .take_while(|block| block.height() >= start_height)
                 .find(|block| block.height() == start_height)
                 .unwrap_or_else(|| {
@@ -397,7 +378,6 @@ impl ConsensusBlockChainImpl {
             consensus_pool,
             consensus_pool.as_block_cache().finalized_chain(),
             tip,
-            None,
         )
         .take_while(|block| block.height() >= start_height)
         .map(|block| (block.height(), block));
@@ -468,7 +448,7 @@ impl ConsensusBlockChainImpl {
         if summary_height >= start_height && summary_height <= tip.height() {
             blocks.insert(summary_height, summary_block.clone());
         }
-        ChainIterator::new(consensus_pool, tip.clone(), None)
+        ChainIterator::new(consensus_pool, tip.clone())
             .take_while(|block| block.height() >= start_height)
             .for_each(|block| {
                 blocks.insert(block.height(), block);

rs/consensus/src/consensus/block_maker.rs

@@ -654,6 +654,7 @@ mod tests {
         signature::ThresholdSignature,
         *,
     };
+    use mockall::Sequence;
     use rstest::rstest;
     use std::sync::Arc;
 
@@ -889,14 +890,26 @@ mod tests {
             pool.insert_validated(summary);
 
             // Payload builder always returns a batch payload with some canister HTTP data
+            let mut sequence = Sequence::new();
             let mut payload_builder = MockPayloadBuilder::new();
-            let expected_payload = BatchPayload {
+            let expected_payload_1 = BatchPayload {
                 canister_http: vec![1; 64],
                 ..Default::default()
             };
+            let expected_payload_2 = BatchPayload {
+                canister_http: vec![2; 64],
+                ..Default::default()
+            };
+            payload_builder
+                .expect_get_payload()
+                .times(1)
+                .return_const(expected_payload_1.clone())
+                .in_sequence(&mut sequence);
             payload_builder
                 .expect_get_payload()
-                .return_const(expected_payload.clone());
+                .times(1)
+                .return_const(expected_payload_2.clone())
+                .in_sequence(&mut sequence);
             let certified_height = Height::from(1);
             state_manager
                 .get_mut()
@@ -952,19 +965,18 @@ mod tests {
             .expect("Block creation should succeed");
             let block = proposal.content.into_inner();
             let filled_batch_payload = &block.payload.as_ref().as_data().batch;
-            assert_eq!(filled_batch_payload, &expected_payload);
+            assert_eq!(filled_batch_payload, &expected_payload_1);
 
-            // Insert the cup at height 10, the batch payload should be empty
-            // Since payloads below the CUP are not returned
+            // Even with the CUP at height 10, the batch payload should include a payload
             pool.insert_validated(cup);
             let proposal = {
                 let reader = PoolReader::new(&pool);
                 block_maker.on_state_change(&reader)
             }
             .expect("Block creation should succeed");
             let block = proposal.content.into_inner();
-            let empty_batch_payload = &block.payload.as_ref().as_data().batch;
-            assert!(empty_batch_payload.is_empty());
+            let filled_batch_payload = &block.payload.as_ref().as_data().batch;
+            assert_eq!(filled_batch_payload, &expected_payload_2);
         });
     }
 

rs/consensus/src/consensus/catchup_package_maker.rs

@@ -172,13 +172,6 @@ impl CatchUpPackageMaker {
         // Skip if random beacon does not exist for the height
         let random_beacon = pool.get_random_beacon(height)?;
 
-        // Skip if the state referenced by finalization tip has not caught up to
-        // this height. This is to increase the chance that states are available to
-        // validate payloads at the chain tip.
-        if pool.get_finalized_tip().context.certified_height < height {
-            return None;
-        }
-
         match self.state_manager.get_state_hash_at(height) {
             Err(StateHashError::Transient(StateNotCommittedYet(_))) => {
                 // TODO: Setup a delay before retry
@@ -336,10 +329,7 @@ mod tests {
             // Skip the first DKG interval
             pool.advance_round_normal_operation_n(interval_length);
 
-            let mut proposal = pool.make_next_block();
-            let block = proposal.content.as_mut();
-            block.context.certified_height = block.height();
-            proposal.content = HashedBlock::new(ic_types::crypto::crypto_hash, block.clone());
+            let proposal = pool.make_next_block();
             pool.insert_validated(proposal.clone());
             pool.notarize(&proposal);
             pool.finalize(&proposal);

rs/consensus/src/consensus/purger.rs

@@ -27,7 +27,7 @@ use ic_interfaces::{
 };
 use ic_interfaces_registry::RegistryClient;
 use ic_interfaces_state_manager::StateManager;
-use ic_logger::{ReplicaLogger, error, trace, warn};
+use ic_logger::{ReplicaLogger, error, info, trace, warn};
 use ic_metrics::MetricsRegistry;
 use ic_replicated_state::ReplicatedState;
 use ic_types::{
@@ -348,6 +348,19 @@ impl Purger {
     /// state removal, check: [`ic_state_manager::StateManagerImpl::remove_states_below`].
     fn purge_checkpoints_below_cup_height(&self, pool: &PoolReader<'_>) {
         let cup_height = pool.get_catch_up_height();
+        let finalized_certified_height = pool.get_finalized_tip().context.certified_height;
+        if finalized_certified_height < cup_height {
+            info!(
+                every_n_seconds => 5,
+                self.log,
+                "Finalized certified height {} is still below the CUP height {}. This \
+                might be caused by a long checkpoint/manifest computation. Skipping \
+                purging checkpoints until finalized certified height catches up.",
+                finalized_certified_height,
+                cup_height
+            );
+            return;
+        }
         self.state_manager.remove_states_below(cup_height);
         trace!(
             self.log,
@@ -742,10 +755,6 @@ mod tests {
 
             let expected_extra_heights = Arc::new(RwLock::new(BTreeSet::new()));
             let extra_heights_clone = Arc::clone(&expected_extra_heights);
-            state_manager
-                .get_mut()
-                .expect_latest_state_height()
-                .return_const(Height::from(0));
             state_manager
                 .get_mut()
                 .expect_update_fast_forward_height()
@@ -814,6 +823,83 @@ mod tests {
         })
     }
 
+    #[test]
+    fn test_purge_checkpoints_below_cup_height() {
+        ic_test_utilities::artifact_pool_config::with_test_pool_config(|pool_config| {
+            let Dependencies {
+                mut pool,
+                state_manager,
+                replica_config,
+                registry,
+                ..
+            } = dependencies(pool_config, 10);
+
+            let purger = Purger::new(
+                replica_config,
+                state_manager.clone(),
+                Arc::new(MockMessageRouting::new()),
+                registry,
+                no_op_logger(),
+                MetricsRegistry::new(),
+            );
+
+            let expected_removed_height = Arc::new(RwLock::new(Height::from(0)));
+            let removed_height_clone = Arc::clone(&expected_removed_height);
+            state_manager
+                .get_mut()
+                .expect_remove_states_below()
+                .times(4) // This will be called exactly four times below
+                .withf(move |height| *height == *removed_height_clone.read().unwrap())
+                .return_const(());
+
+            // Initially, we expect to purge below the genesis CUP of height 0
+            *expected_removed_height.write().unwrap() = Height::from(0);
+            // Expectation called once:
+            purger.purge_checkpoints_below_cup_height(&PoolReader::new(&pool));
+
+            // Advance a bit (less than a DKG interval)
+            pool.advance_round_normal_operation_n(10);
+            // We still expect to purge below the genesis CUP of height 0
+            *expected_removed_height.write().unwrap() = Height::from(0);
+            // Expectation called twice:
+            purger.purge_checkpoints_below_cup_height(&PoolReader::new(&pool));
+
+            // After a DKG interval, we expect to purge below the next CUP height of 60.
+            // THOUGH, because the finalized tip still points to a smaller certified height (of 0
+            // here), the purger should actually not call `state_manager`.
+            pool.advance_round_normal_operation_n(60);
+            // We do not expect the expectation to be called here!
+            purger.purge_checkpoints_below_cup_height(&PoolReader::new(&pool));
+
+            let mut proposal = pool.make_next_block();
+            let block = proposal.content.as_mut();
+            block.context.certified_height = Height::from(60);
+            proposal.content = HashedBlock::new(ic_types::crypto::crypto_hash, block.clone());
+            pool.insert_validated(proposal.clone());
+            pool.notarize(&proposal);
+            pool.finalize(&proposal);
+            // Now that a finalized block points to a height higher or equal (here equal) than the
+            // CUP height, we should purge below the CUP height.
+            *expected_removed_height.write().unwrap() = Height::from(60);
+            // Expectation called three times:
+            purger.purge_checkpoints_below_cup_height(&PoolReader::new(&pool));
+
+            // Same test but with a finalized certified height HIGHER than the CUP height. We also
+            // expect to purge below the CUP height.
+            pool.insert_validated(pool.make_next_beacon());
+            let mut proposal = pool.make_next_block();
+            let block = proposal.content.as_mut();
+            block.context.certified_height = Height::from(62);
+            proposal.content = HashedBlock::new(ic_types::crypto::crypto_hash, block.clone());
+            pool.insert_validated(proposal.clone());
+            pool.notarize(&proposal);
+            pool.finalize(&proposal);
+            *expected_removed_height.write().unwrap() = Height::from(60);
+            // Expectation called four times:
+            purger.purge_checkpoints_below_cup_height(&PoolReader::new(&pool));
+        })
+    }
+
     #[test]
     fn purging_non_finalized_blocks_test() {
         ic_test_utilities::artifact_pool_config::with_test_pool_config(|pool_config| {

rs/consensus/src/consensus/validator.rs

@@ -2973,16 +2973,10 @@ pub mod test {
             let result = validator.check_block_validity(&PoolReader::new(&pool), &test_block);
             assert_matches!(result, Ok(()));
 
-            // Insert the cup at height 10, the payload validation should fail
-            // Since payloads below the CUP are not returned
+            // Even with the CUP at height 10, the validation should still succeed
             pool.insert_validated(cup);
             let result = validator.check_block_validity(&PoolReader::new(&pool), &test_block);
-            assert_matches!(
-                result,
-                Err(ValidationError::ValidationFailed(
-                    ValidationFailure::MissingPastPayloads,
-                ))
-            );
+            assert_matches!(result, Ok(()));
         });
     }