Nomos node tree overlay

consensus-engine/src/overlay/flat_overlay.rs

@@ -1,13 +1,11 @@
+use super::threshold::{apply_threshold, default_super_majority_threshold, deser_fraction};
 use super::LeaderSelection;
 use crate::overlay::CommitteeMembership;
 use crate::{NodeId, Overlay};
-use fraction::{Fraction, ToPrimitive};
+use fraction::Fraction;
 use serde::{Deserialize, Serialize};
 use std::marker::PhantomData;
 
-const LEADER_SUPER_MAJORITY_THRESHOLD_NUM: u64 = 2;
-const LEADER_SUPER_MAJORITY_THRESHOLD_DEN: u64 = 3;
-
 #[derive(Clone, Debug, PartialEq)]
 /// Flat overlay with a single committee and round robin leader selection.
 pub struct FlatOverlay<L: LeaderSelection, M: CommitteeMembership> {
@@ -36,12 +34,8 @@ where
         Self {
             nodes,
             leader,
-            leader_threshold: leader_super_majority_threshold.unwrap_or_else(|| {
-                Fraction::new(
-                    LEADER_SUPER_MAJORITY_THRESHOLD_NUM,
-                    LEADER_SUPER_MAJORITY_THRESHOLD_DEN,
-                )
-            }),
+            leader_threshold: leader_super_majority_threshold
+                .unwrap_or_else(default_super_majority_threshold),
             _committee_membership: Default::default(),
         }
     }
@@ -91,11 +85,7 @@ where
     }
 
     fn leader_super_majority_threshold(&self, _id: NodeId) -> usize {
-        // self.leader_threshold is a tuple of (num, den) where num/den is the super majority threshold
-        (Fraction::from(self.nodes.len()) * self.leader_threshold)
-            .floor()
-            .to_usize()
-            .unwrap()
+        apply_threshold(self.nodes.len(), self.leader_threshold)
     }
 
     fn update_leader_selection<F, E>(&self, f: F) -> Result<Self, E>
@@ -125,30 +115,8 @@ pub struct FlatOverlaySettings<L> {
     pub nodes: Vec<NodeId>,
     /// A fraction representing the threshold in the form `<num>/<den>'
     /// Defaults to 2/3
-    #[serde(with = "deser")]
+    #[serde(with = "deser_fraction")]
     #[serde(skip_serializing_if = "Option::is_none")]
     pub leader_super_majority_threshold: Option<Fraction>,
     pub leader: L,
 }
-
-mod deser {
-    use fraction::Fraction;
-    use serde::{de, Deserialize, Deserializer, Serialize, Serializer};
-    use std::str::FromStr;
-
-    pub fn deserialize<'de, D>(deserializer: D) -> Result<Option<Fraction>, D::Error>
-    where
-        D: Deserializer<'de>,
-    {
-        <Option<String>>::deserialize(deserializer)?
-            .map(|s| FromStr::from_str(&s).map_err(de::Error::custom))
-            .transpose()
-    }
-
-    pub fn serialize<S>(value: &Option<Fraction>, serializer: S) -> Result<S::Ok, S::Error>
-    where
-        S: Serializer,
-    {
-        value.map(|v| v.to_string()).serialize(serializer)
-    }
-}

consensus-engine/src/overlay/mod.rs

@@ -5,6 +5,7 @@ mod flat_overlay;
 mod leadership;
 mod membership;
 mod random_beacon;
+mod threshold;
 mod tree_overlay;
 
 pub use branch_overlay::*;
@@ -98,6 +99,7 @@ mod tests {
             number_of_committees: 1,
             leader: RoundRobin::new(),
             committee_membership: FisherYatesShuffle::new(ENTROPY),
+            super_majority_threshold: None,
         });
         let branch_overlay = BranchOverlay::new(BranchOverlaySettings {
             current_leader: nodes[0],

consensus-engine/src/overlay/threshold.rs

@@ -0,0 +1,38 @@
+use fraction::{Fraction, GenericFraction, ToPrimitive};
+
+const SUPER_MAJORITY_THRESHOLD_NUM: u64 = 2;
+const SUPER_MAJORITY_THRESHOLD_DEN: u64 = 3;
+
+pub(crate) fn default_super_majority_threshold() -> GenericFraction<u64> {
+    Fraction::new(SUPER_MAJORITY_THRESHOLD_NUM, SUPER_MAJORITY_THRESHOLD_DEN)
+}
+
+pub(crate) fn apply_threshold(size: usize, threshold: GenericFraction<u64>) -> usize {
+    // `threshold` is a tuple of (num, den) where `num/den` is the super majority threshold
+    (Fraction::from(size) * threshold)
+        .ceil()
+        .to_usize()
+        .unwrap()
+}
+
+pub mod deser_fraction {
+    use fraction::Fraction;
+    use serde::{de, Deserialize, Deserializer, Serialize, Serializer};
+    use std::str::FromStr;
+
+    pub fn deserialize<'de, D>(deserializer: D) -> Result<Option<Fraction>, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        <Option<String>>::deserialize(deserializer)?
+            .map(|s| FromStr::from_str(&s).map_err(de::Error::custom))
+            .transpose()
+    }
+
+    pub fn serialize<S>(value: &Option<Fraction>, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        value.map(|v| v.to_string()).serialize(serializer)
+    }
+}

consensus-engine/src/overlay/tree_overlay/overlay.rs

@@ -1,14 +1,25 @@
 use super::tree::Tree;
+use crate::overlay::threshold::{
+    apply_threshold, default_super_majority_threshold, deser_fraction,
+};
 use crate::overlay::CommitteeMembership;
 use crate::{overlay::LeaderSelection, Committee, NodeId, Overlay};
+use fraction::Fraction;
+use serde::{Deserialize, Serialize};
 
 #[derive(Debug, Clone)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
 pub struct TreeOverlaySettings<L: LeaderSelection, M: CommitteeMembership> {
     pub nodes: Vec<NodeId>,
     pub current_leader: NodeId,
     pub number_of_committees: usize,
     pub leader: L,
     pub committee_membership: M,
+    /// A fraction representing the threshold in the form `<num>/<den>'
+    /// Defaults to 2/3
+    #[serde(with = "deser_fraction")]
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub super_majority_threshold: Option<Fraction>,
 }
 
 #[derive(Debug, Clone)]
@@ -19,6 +30,7 @@ pub struct TreeOverlay<L, M> {
     pub(super) carnot_tree: Tree,
     pub(super) leader: L,
     pub(super) committee_membership: M,
+    pub(super) threshold: Fraction,
 }
 
 impl<L, M> Overlay for TreeOverlay<L, M>
@@ -38,6 +50,7 @@ where
             number_of_committees,
             leader,
             committee_membership,
+            super_majority_threshold,
         } = settings;
 
         committee_membership.reshape_committees(&mut nodes);
@@ -50,6 +63,7 @@ where
             carnot_tree,
             leader,
             committee_membership,
+            threshold: super_majority_threshold.unwrap_or_else(default_super_majority_threshold),
         }
     }
 
@@ -134,7 +148,7 @@ where
         }
         self.carnot_tree
             .committee_by_member_id(&id)
-            .map(|c| (c.len() * 2 / 3) + 1)
+            .map(|c| apply_threshold(c.len(), self.threshold))
             .expect("node is not part of any committee")
     }
 
@@ -156,8 +170,7 @@ where
         // });
         // let root_size = self.root_committee().len();
         // let committee_size = root_size + children_size;
-        let committee_size = self.root_committee().len();
-        (committee_size * 2 / 3) + 1
+        apply_threshold(self.root_committee().len(), self.threshold)
     }
 
     fn update_leader_selection<F, E>(&self, f: F) -> Result<Self, E>
@@ -184,6 +197,7 @@ where
                 number_of_committees: self.number_of_committees,
                 leader: self.leader.clone(),
                 committee_membership,
+                super_majority_threshold: Some(self.threshold),
             };
             Self::new(settings)
         })
@@ -202,6 +216,7 @@ where
             number_of_committees: self.number_of_committees,
             leader,
             committee_membership,
+            super_majority_threshold: Some(self.threshold),
         })
     }
 
@@ -232,6 +247,7 @@ mod tests {
             number_of_committees: 3,
             leader: RoundRobin::new(),
             committee_membership: FisherYatesShuffle::new(ENTROPY),
+            super_majority_threshold: None,
         });
 
         assert_eq!(*overlay.leader(), nodes[0]);
@@ -246,6 +262,7 @@ mod tests {
             number_of_committees: 3,
             leader: RoundRobin::new(),
             committee_membership: FisherYatesShuffle::new(ENTROPY),
+            super_majority_threshold: None,
         });
 
         let leader = overlay.next_leader();
@@ -263,6 +280,7 @@ mod tests {
             number_of_committees: 3,
             leader: RoundRobin::new(),
             committee_membership: FisherYatesShuffle::new(ENTROPY),
+            super_majority_threshold: None,
         });
 
         let mut expected_root = Committee::new();
@@ -281,6 +299,7 @@ mod tests {
             number_of_committees: 3,
             leader: RoundRobin::new(),
             committee_membership: FisherYatesShuffle::new(ENTROPY),
+            super_majority_threshold: None,
         });
 
         let mut leaf_committees = overlay
@@ -311,6 +330,7 @@ mod tests {
             number_of_committees: 3,
             leader: RoundRobin::new(),
             committee_membership: FisherYatesShuffle::new(ENTROPY),
+            super_majority_threshold: None,
         });
 
         assert_eq!(overlay.super_majority_threshold(overlay.nodes[8]), 0);
@@ -325,6 +345,7 @@ mod tests {
             number_of_committees: 3,
             leader: RoundRobin::new(),
             committee_membership: FisherYatesShuffle::new(ENTROPY),
+            super_majority_threshold: None,
         });
 
         assert_eq!(overlay.super_majority_threshold(overlay.nodes[0]), 3);
@@ -339,6 +360,7 @@ mod tests {
             number_of_committees: 3,
             leader: RoundRobin::new(),
             committee_membership: FisherYatesShuffle::new(ENTROPY),
+            super_majority_threshold: None,
         });
 
         assert_eq!(

nodes/nomos-node/src/config.rs

@@ -87,30 +87,19 @@ pub struct ConsensusArgs {
     consensus_timeout_secs: Option<String>,
 }
 
-#[derive(ValueEnum, Clone, Debug, Default)]
-pub enum OverlayType {
-    #[default]
-    Flat,
-    Tree,
-}
-
 #[derive(Parser, Debug, Clone)]
 pub struct OverlayArgs {
-    // TODO: Act on type and support other overlays.
-    #[clap(long = "overlay-type", env = "OVERLAY_TYPE")]
-    pub overlay_type: Option<OverlayType>,
-
     #[clap(long = "overlay-nodes", env = "OVERLAY_NODES", num_args = 1.., value_delimiter = ',')]
     pub overlay_nodes: Option<Vec<String>>,
 
     #[clap(long = "overlay-leader", env = "OVERLAY_LEADER")]
-    pub overlay_leader: Option<usize>,
+    pub overlay_leader: Option<String>,
 
     #[clap(
         long = "overlay-leader-super-majority-threshold",
-        env = "OVERLAY_LEADER_SUPER_MAJORITY_THRESHOLD"
+        env = "OVERLAY_NUMBER_OF_COMMITTEES"
     )]
-    pub overlay_leader_super_majority_threshold: Option<usize>,
+    pub overlay_number_of_committees: Option<usize>,
 }
 
 #[derive(Deserialize, Debug, Clone, Serialize)]
@@ -237,8 +226,8 @@ impl Config {
     pub fn update_overlay(mut self, overlay_args: OverlayArgs) -> Result<Self> {
         let OverlayArgs {
             overlay_nodes,
-            overlay_leader_super_majority_threshold,
-            ..
+            overlay_leader,
+            overlay_number_of_committees,
         } = overlay_args;
 
         if let Some(nodes) = overlay_nodes {
@@ -252,10 +241,13 @@ impl Config {
                 .collect::<Result<Vec<_>, eyre::Report>>()?;
         }
 
-        if let Some(threshold) = overlay_leader_super_majority_threshold {
-            self.consensus
-                .overlay_settings
-                .leader_super_majority_threshold = Some(threshold.into());
+        if let Some(leader) = overlay_leader {
+            let bytes = <[u8; 32]>::from_hex(leader)?;
+            self.consensus.overlay_settings.current_leader = bytes.into();
+        }
+
+        if let Some(committees) = overlay_number_of_committees {
+            self.consensus.overlay_settings.number_of_committees = committees;
         }
 
         Ok(self)

nodes/nomos-node/src/lib.rs

@@ -2,7 +2,7 @@ mod config;
 mod tx;
 
 use color_eyre::eyre::Result;
-use consensus_engine::overlay::{FlatOverlay, RandomBeaconState, RoundRobin};
+use consensus_engine::overlay::{RandomBeaconState, RoundRobin, TreeOverlay};
 use full_replication::Certificate;
 use full_replication::{AbsoluteNumber, Attestation, Blob, FullReplication};
 #[cfg(feature = "metrics")]
@@ -53,7 +53,7 @@ pub type Carnot = CarnotConsensus<
         Certificate,
         <<Certificate as certificate::Certificate>::Blob as blob::Blob>::Hash,
     >,
-    FlatOverlay<RoundRobin, RandomBeaconState>,
+    TreeOverlay<RoundRobin, RandomBeaconState>,
     FillSizeWithTx<MB16, Tx>,
     FillSizeWithBlobsCertificate<MB16, Certificate>,
 >;

nomos-services/consensus/src/lib.rs

@@ -292,7 +292,8 @@ where
                 )
                 .await
                 else {
-                    unreachable!()
+                    tracing::debug!("Failed to gather initial votes");
+                    return Event::None;
                 };
                 Event::ProposeBlock { qc }
             });
@@ -782,8 +783,9 @@ where
         let votes_stream = adapter.votes_stream(&committee, block.view, block.id).await;
         match tally.tally(block.clone(), votes_stream).await {
             Ok((qc, votes)) => Event::Approve { qc, votes, block },
-            Err(_e) => {
-                todo!("Handle tally error {_e}");
+            Err(e) => {
+                tracing::debug!("Error gathering votes: {e}");
+                Event::None
             }
         }
     }

nomos-services/consensus/src/tally/happy.rs

@@ -20,6 +20,8 @@ pub enum CarnotTallyError {
     InvalidVote(String),
     #[error("Did not receive enough votes")]
     InsufficientVotes,
+    #[error("The vote stream ended without tally")]
+    StreamEnded,
 }
 
 #[derive(Clone, Debug)]
@@ -80,6 +82,7 @@ impl Tally for CarnotTally {
                 ));
             }
         }
-        unreachable!()
+
+        Err(CarnotTallyError::StreamEnded)
     }
 }