review feedback

fendermint/vm/topdown/src/observation.rs

@@ -10,7 +10,7 @@ use fendermint_vm_genesis::ValidatorKey;
 use serde::{Deserialize, Serialize};
 use std::fmt::{Display, Formatter};
 
-/// The content that validators gossip among each other.
+/// The parent subnet data fetch through RPC endpoint
 #[derive(Serialize, Deserialize, Hash, Debug, Clone, Eq, PartialEq, Arbitrary)]
 pub struct Observation {
     pub(crate) parent_height: u64,
@@ -62,7 +62,11 @@ impl CertifiedObservation {
         let (pk2, _) = self.signature.recover(p.as_slice())?;
 
         if pk1 != pk2 {
-            return Err(anyhow!("public keys not aligned"));
+            return Err(anyhow!(
+                "public keys not aligned {}, {}",
+                hex::encode(pk1.serialize()),
+                hex::encode(pk2.serialize())
+            ));
         }
 
         Ok(ValidatorKey::new(pk1))

fendermint/vm/topdown/src/vote/mod.rs

@@ -231,11 +231,7 @@ where
                 let _ = tx.send(());
             }
             VoteReactorRequest::FindQuorum(tx) => {
-                let quorum = self
-                    .vote_tally
-                    .find_quorum()
-                    .inspect_err(|e| tracing::error!(err = e.to_string(), "cannot find quorum"))
-                    .unwrap_or_default();
+                let quorum = self.vote_tally.get_latest_quorum();
                 let _ = tx.send(quorum);
             }
             VoteReactorRequest::SetPowerTable { updates, tx } => {

fendermint/vm/topdown/src/vote/store.rs

@@ -118,6 +118,30 @@ impl<'a> VoteAgg<'a> {
 
         votes
     }
+
+    pub fn max_observation_weight(
+        &self,
+        power_table: &PowerTable,
+    ) -> Option<(&Observation, Weight)> {
+        let mut votes: HashMap<&Observation, Weight> = HashMap::new();
+
+        for v in self.0.iter() {
+            let validator = v.voter();
+
+            let power = power_table.get(&validator).cloned().unwrap_or(0);
+            if power == 0 {
+                continue;
+            }
+
+            let ob = v.observation();
+            votes.entry(ob).and_modify(|w| *w += power).or_insert(power);
+        }
+
+        votes
+            .iter()
+            .max_by(|a, b| a.1.cmp(b.1))
+            .map(|(k, v)| (*k, *v))
+    }
 }
 
 #[cfg(test)]
@@ -181,7 +205,10 @@ mod tests {
         );
 
         let agg = VoteAgg(votes.iter().collect());
-        let weights = agg.observation_weights(&HashMap::from_iter(powers));
-        assert_eq!(weights, vec![(&observation1, 1), (&observation2, 2),])
+        let (ob, weight) = agg
+            .max_observation_weight(&HashMap::from_iter(powers))
+            .unwrap();
+        assert_eq!(ob, &observation2);
+        assert_eq!(weight, 2);
     }
 }

fendermint/vm/topdown/src/vote/tally.rs

@@ -24,6 +24,9 @@ pub(crate) struct VoteTally<S> {
 
     /// The latest height that was voted to be finalized and committed to child blockchian
     last_finalized_height: BlockHeight,
+
+    /// The latest observation with a quorum formed
+    latest_quorum_formed: Option<Observation>,
 }
 
 impl<S: VoteStore> VoteTally<S> {
@@ -40,11 +43,16 @@ impl<S: VoteStore> VoteTally<S> {
                 store.purge_votes_at_height(i)?;
             }
         }
-        Ok(Self {
+
+        let mut s = Self {
             power_table: HashMap::from_iter(power_table),
             votes: store,
             last_finalized_height,
-        })
+            latest_quorum_formed: None,
+        };
+        s.latest_quorum_formed = s.find_quorum()?;
+
+        Ok(s)
     }
 
     /// Check that a validator key is currently part of the power table.
@@ -139,11 +147,37 @@ impl<S: VoteStore> VoteTally<S> {
 
         self.votes.store_vote(parent_height, vote)?;
 
+        // check if a new quorum is formed with the new vote
+        let is_check_quorum = self
+            .latest_quorum_formed
+            .as_ref()
+            // we only check quorum if the incoming vote height is bigger than latest quorum formed,
+            // no point choosing a lower parent height
+            .map(|v| v.parent_height < parent_height)
+            // if there is quorum formed, trigger finding quorum
+            .unwrap_or(true);
+        if !is_check_quorum {
+            return Ok(true);
+        }
+
+        if let Some(ob) = self.check_quorum_at_height(parent_height)? {
+            self.latest_quorum_formed = Some(ob);
+        }
+
         Ok(true)
     }
 
+    pub fn get_latest_quorum(&mut self) -> Option<Observation> {
+        if let Some(ref v) = self.latest_quorum_formed {
+            if self.last_finalized_height < v.parent_height {
+                return Some(v.clone());
+            }
+        }
+        None
+    }
+
     /// Find a block on the (from our perspective) finalized chain that gathered enough votes from validators.
-    pub fn find_quorum(&self) -> Result<Option<Observation>, Error> {
+    fn find_quorum(&self) -> Result<Option<Observation>, Error> {
         let quorum_threshold = self.quorum_threshold();
         let Some(max_height) = self.votes.latest_vote_height()? else {
             tracing::info!("vote store has no vote yet, skip finding quorum");
@@ -179,6 +213,7 @@ impl<S: VoteStore> VoteTally<S> {
     pub fn set_finalized(&mut self, block_height: BlockHeight) -> Result<(), Error> {
         self.votes.purge_votes_at_height(block_height)?;
         self.last_finalized_height = block_height;
+        self.latest_quorum_formed = None;
         Ok(())
     }
 
@@ -209,6 +244,30 @@ impl<S: VoteStore> VoteTally<S> {
             }
         }
     }
+
+    fn check_quorum_at_height(
+        &self,
+        block_height: BlockHeight,
+    ) -> Result<Option<Observation>, Error> {
+        let quorum_threshold = self.quorum_threshold();
+        let votes = self.votes.get_votes_at_height(block_height)?;
+
+        if let Some((observation, weight)) = votes.max_observation_weight(&self.power_table) {
+            tracing::info!(
+                height = block_height,
+                observation = observation.to_string(),
+                weight,
+                quorum_threshold,
+                "observation and weight"
+            );
+
+            if weight >= quorum_threshold {
+                return Ok(Some(observation.clone()));
+            }
+        }
+
+        Ok(None)
+    }
 }
 
 #[cfg(test)]