feat: no state hashing during catch-up (state manager)

Author: mraszyk

rs/interfaces/state_manager/mocks/src/lib.rs

@@ -51,6 +51,8 @@ mock! {
 
         fn list_state_hashes_to_certify(&self) -> Vec<(Height, CryptoHashOfPartialState)>;
 
+        fn list_state_heights_to_certify(&self) -> Vec<Height>;
+
         fn deliver_state_certification(&self, certification: Certification);
 
         fn remove_states_below(&self, height: Height);

rs/interfaces/state_manager/src/lib.rs

@@ -146,6 +146,10 @@ pub trait StateManager: StateReader {
     ///   ```
     fn list_state_hashes_to_certify(&self) -> Vec<(Height, CryptoHashOfPartialState)>;
 
+    /// Returns a list of heights for which the state manager optimistically requests
+    /// a certification to be delivered via `state_manager.deliver_state_certification`.
+    fn list_state_heights_to_certify(&self) -> Vec<Height>;
+
     /// Delivers a `certification` corresponding to some state hash / height
     /// pair.
     ///

rs/state_machine_tests/src/lib.rs

@@ -942,6 +942,10 @@ impl StateManager for StateMachineStateManager {
         self.deref().list_state_hashes_to_certify()
     }
 
+    fn list_state_heights_to_certify(&self) -> Vec<Height> {
+        self.deref().list_state_heights_to_certify()
+    }
+
     fn deliver_state_certification(&self, certification: Certification) {
         self.deref().deliver_state_certification(certification)
     }

rs/state_manager/src/lib.rs

@@ -83,7 +83,7 @@ use std::sync::{
 };
 use std::time::{Duration, Instant, SystemTime};
 use std::{
-    collections::{BTreeMap, BTreeSet, VecDeque},
+    collections::{BTreeMap, BTreeSet, HashSet, VecDeque},
     sync::Mutex,
 };
 use tempfile::tempfile;
@@ -118,9 +118,15 @@ const CRITICAL_ERROR_REPLICATED_STATE_ALTERED_AFTER_CHECKPOINT: &str =
 const ARCHIVED_DIVERGED_CHECKPOINT_MAX_AGE: Duration = Duration::from_secs(30 * 24 * 60 * 60); // 30 days
 
 /// The maximum number of consecutive rounds for which the optimization of
-/// skipping state cloning and certification metadata computation triggers.
+/// skipping state cloning and computing certification metadata triggers
+/// while catching up.
 const MAX_CONSECUTIVE_ROUNDS_WITHOUT_STATE_CLONING: u64 = 10;
 
+/// The maximum number of future heights starting at tip height
+/// that the state manager optimistically asks consensus to certify
+/// in advance while catching up.
+const MAX_FUTURE_HEIGHTS_TO_CERTIFY: u64 = 20;
+
 /// Write an overlay file this many rounds before each checkpoint.
 pub const NUM_ROUNDS_BEFORE_CHECKPOINT_TO_WRITE_OVERLAY: u64 = 50;
 
@@ -765,6 +771,7 @@ impl StateSyncMetrics {
 type StatesMetadata = BTreeMap<Height, StateMetadata>;
 
 type CertificationsMetadata = BTreeMap<Height, CertificationMetadata>;
+type Certifications = BTreeMap<Height, Certification>;
 
 /// This struct bundles the root hash, manifest and meta-manifest.
 #[derive(Clone, Debug)]
@@ -886,6 +893,8 @@ impl StateSyncRefs {
 struct SharedState {
     /// Certifications metadata kept for all states
     certifications_metadata: CertificationsMetadata,
+    /// Certifications delivered optimistically to optimize state hashing.
+    certifications: Certifications,
     /// Metadata for each checkpoint
     states_metadata: StatesMetadata,
     /// A list of states present in the memory.  This list is guaranteed to not be
@@ -1556,6 +1565,7 @@ impl StateManagerImpl {
 
         let states = Arc::new(parking_lot::RwLock::new(SharedState {
             certifications_metadata,
+            certifications: BTreeMap::new(),
             states_metadata,
             snapshots,
             last_advertised: Self::INITIAL_STATE_HEIGHT,
@@ -2316,6 +2326,10 @@ impl StateManagerImpl {
             .latest_certified_height
             .set(latest_certified_height.get() as i64);
 
+        let mut certifications = states.certifications.split_off(&last_height_to_keep);
+        std::mem::swap(&mut certifications, &mut states.certifications);
+        self.deallocator_thread.send(Box::new(certifications));
+
         let mut metadata_to_keep = states.states_metadata.split_off(&last_height_to_keep);
 
         for h in checkpoint_heights_to_keep.iter() {
@@ -2711,6 +2725,8 @@ impl StateManager for StateManagerImpl {
                     states.certifications_metadata.get(&tip_height)
                 {
                     CryptoHashOfPartialState::from(tip_metadata.certified_state_hash.clone())
+                } else if let Some(tip_certification) = states.certifications.get(&tip_height) {
+                    tip_certification.signed.content.hash.clone()
                 } else {
                     std::mem::drop(states);
 
@@ -2987,6 +3003,27 @@ impl StateManager for StateManagerImpl {
             .collect()
     }
 
+    fn list_state_heights_to_certify(&self) -> Vec<Height> {
+        let states = self.states.read();
+        let tip_height = states.tip_height.get();
+        let heights_with_certification: HashSet<_> =
+            states.certifications.keys().cloned().collect();
+        drop(states);
+
+        let latest_subnet_certified_height =
+            self.latest_subnet_certified_height.load(Ordering::Relaxed);
+        let state_heights = tip_height
+            ..min(
+                tip_height + MAX_FUTURE_HEIGHTS_TO_CERTIFY,
+                latest_subnet_certified_height,
+            );
+        state_heights
+            .into_iter()
+            .map(Height::new)
+            .filter(|h| !heights_with_certification.contains(h))
+            .collect()
+    }
+
     fn deliver_state_certification(&self, certification: Certification) {
         let _timer = self
             .metrics
@@ -3038,6 +3075,9 @@ impl StateManager for StateManagerImpl {
                     self.deallocator_thread.send(Box::new(tree));
                 }
             }
+        } else {
+            let height = certification.height;
+            states.certifications.insert(height, certification);
         }
     }
 
@@ -3337,11 +3377,7 @@ impl StateManager for StateManagerImpl {
 
         assert_tip_is_none(&states);
 
-        // It's possible that we already computed this state before.  We
-        // validate that hashes agree to spot bugs causing non-determinism as
-        // early as possible.
-        if let Some(prev_metadata) = states.certifications_metadata.get(&height) {
-            let prev_hash = &prev_metadata.certified_state_hash;
+        let assert_prev_hash_matches = |prev_hash| {
             let hash = &certification_metadata.certified_state_hash;
             if prev_hash != hash {
                 if let Err(err) = self.state_layout.create_diverged_state_marker(height) {
@@ -3354,6 +3390,23 @@ impl StateManager for StateManagerImpl {
                     "Committed state @{height} with hash {hash:?} which is different from previously computed or delivered hash {prev_hash:?}"
                 );
             }
+        };
+
+        // It's possible that we already computed this state before.  We
+        // validate that hashes agree to spot bugs causing non-determinism as
+        // early as possible.
+        if let Some(prev_metadata) = states.certifications_metadata.get(&height) {
+            let prev_hash = &prev_metadata.certified_state_hash;
+            assert_prev_hash_matches(prev_hash);
+        }
+
+        // We reuse certification delivered by consensus if possible.
+        // We also validate that hashes agree to spot bugs causing non-determinism as
+        // early as possible.
+        if let Some(certification) = states.certifications.get(&height) {
+            let prev_hash = &certification.signed.content.hash.clone().get();
+            assert_prev_hash_matches(prev_hash);
+            certification_metadata.certification = Some(certification.clone());
         }
 
         if !states
@@ -4093,6 +4146,9 @@ pub mod testing {
         /// Testing only: Returns certification at a given height in `states.certifications_metadata`.
         fn certifications_metadata_certification(&self, height: Height) -> Option<Certification>;
 
+        /// Testing only: Returns certifications in `states.certifications`.
+        fn certifications(&self) -> BTreeMap<Height, Certification>;
+
         /// Testing only: Returns `fast_forward_height`.
         fn fast_forward_height(&self) -> u64;
     }
@@ -4175,6 +4231,11 @@ pub mod testing {
                 .clone()
         }
 
+        fn certifications(&self) -> BTreeMap<Height, Certification> {
+            let states = self.states.read();
+            states.certifications.clone()
+        }
+
         fn fast_forward_height(&self) -> u64 {
             self.fast_forward_height.load(Ordering::Relaxed)
         }

rs/state_manager/src/tip.rs

@@ -1679,6 +1679,7 @@ mod test {
 
             let dummy_states = Arc::new(parking_lot::RwLock::new(SharedState {
                 certifications_metadata: Default::default(),
+                certifications: Default::default(),
                 states_metadata: Default::default(),
                 snapshots: Default::default(),
                 last_advertised: Height::new(0),