Nomos node tree overlay (#415)

* Use tree overlay in nomos node

* Use tree overlay in tests module

* Handle unexpected consensus vote stream end in tally

* Spawn the next leader first (#425)

* Report unhappy blocks in happy path test (#430)

* report unhappy blocks in the happy path test

* Make threshold configurable for TreeOverlay (#426)

* Modified test, so that all nodes don’t all connect to the first node only (#442)

* merge fix

---------

Co-authored-by: Youngjoon Lee <taxihighway@gmail.com>
Co-authored-by: Al Liu <scygliu1@gmail.com>

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(|| {
+            leader_threshold: leader_super_majority_threshold
-                Fraction::new(
+                .unwrap_or_else(default_super_majority_threshold),
-                    LEADER_SUPER_MAJORITY_THRESHOLD_NUM,
-                    LEADER_SUPER_MAJORITY_THRESHOLD_DEN,
-                )
-            }),
             _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
+        apply_threshold(self.nodes.len(), self.leader_threshold)
-        (Fraction::from(self.nodes.len()) * self.leader_threshold)
-            .floor()
-            .to_usize()
-            .unwrap()
     }
 
     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();
+        apply_threshold(self.root_committee().len(), self.threshold)
-        (committee_size * 2 / 3) + 1
     }
 
     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 {
+        if let Some(leader) = overlay_leader {
-            self.consensus
+            let bytes = <[u8; 32]>::from_hex(leader)?;
-                .overlay_settings
+            self.consensus.overlay_settings.current_leader = bytes.into();
-                .leader_super_majority_threshold = Some(threshold.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) => {
+            Err(e) => {
-                todo!("Handle tally error {_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)
     }
 }

simulations/src/bin/app/overlay_node.rs

@@ -57,6 +57,7 @@ pub fn to_overlay_node<R: Rng>(
                 number_of_committees: tree_settings.number_of_committees,
                 leader: RoundRobin::new(),
                 committee_membership: RandomBeaconState::initial_sad_from_entropy([0; 32]),
+                super_majority_threshold: None,
             };
             Box::new(
                 CarnotNode::<TreeOverlay<RoundRobin, RandomBeaconState>>::new(

tests/Cargo.toml

@@ -27,6 +27,7 @@ secp256k1 = { version = "0.26", features = ["rand"] }
 reqwest = { version = "0.11", features = ["json"] }
 nomos-libp2p = { path = "../nomos-libp2p" }
 tempfile = "3.6"
+serde = { version = "1", features = ["derive"] }
 serde_yaml = "0.9"
 serde_json = "1.0"
 tokio = "1"

tests/src/lib.rs

@@ -35,11 +35,27 @@ pub trait Node: Sized {
 
 #[derive(Clone)]
 pub enum SpawnConfig {
+    // Star topology: Every node is initially connected to a single node.
     Star {
-        n_participants: usize,
+        consensus: ConsensusConfig,
-        threshold: Fraction,
+        mixnet: MixnetConfig,
-        timeout: Duration,
+    },
-        mixnet_node_configs: Vec<MixnetNodeConfig>,
+    // Chain topology: Every node is chained to the node next to it.
-        mixnet_topology: MixnetTopology,
+    Chain {
+        consensus: ConsensusConfig,
+        mixnet: MixnetConfig,
     },
 }
+
+#[derive(Clone)]
+pub struct ConsensusConfig {
+    pub n_participants: usize,
+    pub threshold: Fraction,
+    pub timeout: Duration,
+}
+
+#[derive(Clone)]
+pub struct MixnetConfig {
+    pub node_configs: Vec<MixnetNodeConfig>,
+    pub topology: MixnetTopology,
+}

tests/src/nodes/mixnode.rs

@@ -9,7 +9,7 @@ use mixnet_topology::{Layer, MixnetTopology, Node};
 use rand::{thread_rng, RngCore};
 use tempfile::NamedTempFile;
 
-use crate::get_available_port;
+use crate::{get_available_port, MixnetConfig};
 
 const MIXNODE_BIN: &str = "../target/debug/mixnode";
 
@@ -46,9 +46,7 @@ impl MixNode {
         Self { child }
     }
 
-    pub async fn spawn_nodes(
+    pub async fn spawn_nodes(num_nodes: usize) -> (Vec<Self>, MixnetConfig) {
-        num_nodes: usize,
-    ) -> (Vec<Self>, Vec<MixnetNodeConfig>, MixnetTopology) {
         let mut configs = Vec::<MixnetNodeConfig>::new();
         for _ in 0..num_nodes {
             let mut private_key = [0u8; PRIVATE_KEY_SIZE];
@@ -76,7 +74,13 @@ impl MixNode {
         }
 
         // We need to return configs as well, to configure mixclients accordingly
-        (nodes, configs.clone(), Self::build_topology(configs))
+        (
+            nodes,
+            MixnetConfig {
+                node_configs: configs.clone(),
+                topology: Self::build_topology(configs),
+            },
+        )
     }
 
     fn build_topology(configs: Vec<MixnetNodeConfig>) -> MixnetTopology {

tests/src/nodes/nomos.rs

@@ -3,18 +3,18 @@ use std::net::SocketAddr;
 use std::process::{Child, Command, Stdio};
 use std::time::Duration;
 // internal
-use crate::{get_available_port, Node, SpawnConfig};
+use crate::{get_available_port, ConsensusConfig, MixnetConfig, Node, SpawnConfig};
-use consensus_engine::overlay::{FlatOverlaySettings, RoundRobin};
+use consensus_engine::overlay::{RandomBeaconState, RoundRobin, TreeOverlay, TreeOverlaySettings};
-use consensus_engine::NodeId;
+use consensus_engine::{NodeId, Overlay};
 use mixnet_client::{MixnetClientConfig, MixnetClientMode};
 use mixnet_node::MixnetNodeConfig;
 use mixnet_topology::MixnetTopology;
 use nomos_consensus::{CarnotInfo, CarnotSettings};
 use nomos_http::backends::axum::AxumBackendSettings;
-use nomos_libp2p::Multiaddr;
+use nomos_libp2p::{multiaddr, Multiaddr};
 use nomos_log::{LoggerBackend, LoggerFormat};
 use nomos_mempool::MempoolMetrics;
-use nomos_network::backends::libp2p::{Libp2pConfig, Libp2pInfo};
+use nomos_network::backends::libp2p::Libp2pConfig;
 use nomos_network::NetworkConfig;
 use nomos_node::Config;
 // crates
@@ -27,7 +27,6 @@ use tempfile::NamedTempFile;
 static CLIENT: Lazy<Client> = Lazy::new(Client::new);
 const NOMOS_BIN: &str = "../target/debug/nomos-node";
 const CARNOT_INFO_API: &str = "carnot/info";
-const NETWORK_INFO_API: &str = "network/info";
 const MEMPOOL_API: &str = "mempool-";
 const LOGS_PREFIX: &str = "__logs";
 
@@ -101,16 +100,6 @@ impl NomosNode {
             tokio::time::sleep(Duration::from_millis(100)).await;
         }
     }
-    pub async fn get_listening_address(&self) -> Multiaddr {
-        self.get(NETWORK_INFO_API)
-            .await
-            .unwrap()
-            .json::<Libp2pInfo>()
-            .await
-            .unwrap()
-            .listen_addresses
-            .swap_remove(0)
-    }
 
     pub async fn get_mempoool_metrics(&self, pool: Pool) -> MempoolMetrics {
         let discr = match pool {
@@ -164,37 +153,29 @@ impl Node for NomosNode {
 
     async fn spawn_nodes(config: SpawnConfig) -> Vec<Self> {
         match config {
-            SpawnConfig::Star {
+            SpawnConfig::Star { consensus, mixnet } => {
-                n_participants,
+                let (next_leader_config, configs) = create_node_configs(consensus, mixnet);
-                threshold,
+
-                timeout,
+                let first_node_addr = node_address(&next_leader_config);
-                mut mixnet_node_configs,
+                let mut nodes = vec![Self::spawn(next_leader_config).await];
-                mixnet_topology,
-            } => {
-                let mut ids = vec![[0; 32]; n_participants];
-                for id in &mut ids {
-                    thread_rng().fill(id);
-                }
-                let mut configs = ids
-                    .iter()
-                    .map(|id| {
-                        create_node_config(
-                            ids.iter().copied().map(NodeId::new).collect(),
-                            *id,
-                            threshold,
-                            timeout,
-                            mixnet_node_configs.pop(),
-                            mixnet_topology.clone(),
-                        )
-                    })
-                    .collect::<Vec<_>>();
-                let mut nodes = vec![Self::spawn(configs.swap_remove(0)).await];
-                let listening_addr = nodes[0].get_listening_address().await;
                 for mut conf in configs {
                     conf.network
                         .backend
                         .initial_peers
-                        .push(listening_addr.clone());
+                        .push(first_node_addr.clone());
+
+                    nodes.push(Self::spawn(conf).await);
+                }
+                nodes
+            }
+            SpawnConfig::Chain { consensus, mixnet } => {
+                let (next_leader_config, configs) = create_node_configs(consensus, mixnet);
+
+                let mut prev_node_addr = node_address(&next_leader_config);
+                let mut nodes = vec![Self::spawn(next_leader_config).await];
+                for mut conf in configs {
+                    conf.network.backend.initial_peers.push(prev_node_addr);
+                    prev_node_addr = node_address(&conf);
 
                     nodes.push(Self::spawn(conf).await);
                 }
@@ -217,6 +198,47 @@ impl Node for NomosNode {
     }
 }
 
+/// Returns the config of the next leader and all other nodes.
+///
+/// Depending on the network topology, the next leader must be spawned first,
+/// so the leader can receive votes from all other nodes that will be subsequently spawned.
+/// If not, the leader will miss votes from nodes spawned before itself.
+/// This issue will be resolved by devising the block catch-up mechanism in the future.
+fn create_node_configs(
+    consensus: ConsensusConfig,
+    mut mixnet: MixnetConfig,
+) -> (Config, Vec<Config>) {
+    let mut ids = vec![[0; 32]; consensus.n_participants];
+    for id in &mut ids {
+        thread_rng().fill(id);
+    }
+
+    let mut configs = ids
+        .iter()
+        .map(|id| {
+            create_node_config(
+                ids.iter().copied().map(NodeId::new).collect(),
+                *id,
+                consensus.threshold,
+                consensus.timeout,
+                mixnet.node_configs.pop(),
+                mixnet.topology.clone(),
+            )
+        })
+        .collect::<Vec<_>>();
+
+    let overlay = TreeOverlay::new(configs[0].consensus.overlay_settings.clone());
+    let next_leader = overlay.next_leader();
+    let next_leader_idx = ids
+        .iter()
+        .position(|&id| NodeId::from(id) == next_leader)
+        .unwrap();
+
+    let next_leader_config = configs.swap_remove(next_leader_idx);
+
+    (next_leader_config, configs)
+}
+
 fn create_node_config(
     nodes: Vec<NodeId>,
     private_key: [u8; 32],
@@ -247,13 +269,16 @@ fn create_node_config(
         },
         consensus: CarnotSettings {
             private_key,
-            overlay_settings: FlatOverlaySettings {
+            overlay_settings: TreeOverlaySettings {
                 nodes,
                 leader: RoundRobin::new(),
-                // By setting the leader_threshold to 1 we ensure that all nodes come
+                current_leader: [0; 32].into(),
+                number_of_committees: 1,
+                committee_membership: RandomBeaconState::initial_sad_from_entropy([0; 32]),
+                // By setting the threshold to 1 we ensure that all nodes come
                 // online before progressing. This is only necessary until we add a way
                 // to recover poast blocks from other nodes.
-                leader_super_majority_threshold: Some(threshold),
+                super_majority_threshold: Some(threshold),
             },
             timeout,
             transaction_selector_settings: (),
@@ -286,6 +311,10 @@ fn create_node_config(
     config
 }
 
+fn node_address(config: &Config) -> Multiaddr {
+    multiaddr!(Ip4([127, 0, 0, 1]), Tcp(config.network.backend.inner.port))
+}
+
 pub enum Pool {
     Da,
     Cl,

tests/src/tests/cli.rs

@@ -7,19 +7,20 @@ use nomos_cli::cmds::{
 };
 use std::time::Duration;
 use tempfile::NamedTempFile;
-use tests::{nodes::nomos::Pool, MixNode, Node, NomosNode, SpawnConfig};
+use tests::{nodes::nomos::Pool, ConsensusConfig, MixNode, Node, NomosNode, SpawnConfig};
 
 const TIMEOUT_SECS: u64 = 20;
 
 #[tokio::test]
 async fn disseminate_blob() {
-    let (_mixnodes, mixnet_node_configs, mixnet_topology) = MixNode::spawn_nodes(2).await;
+    let (_mixnodes, mixnet_config) = MixNode::spawn_nodes(2).await;
-    let mut nodes = NomosNode::spawn_nodes(SpawnConfig::Star {
+    let mut nodes = NomosNode::spawn_nodes(SpawnConfig::Chain {
+        consensus: ConsensusConfig {
             n_participants: 2,
             threshold: Fraction::one(),
             timeout: Duration::from_secs(10),
-        mixnet_node_configs,
+        },
-        mixnet_topology,
+        mixnet: mixnet_config,
     })
     .await;
 

tests/src/tests/happy.rs

@@ -1,12 +1,19 @@
-use consensus_engine::View;
+use consensus_engine::{Qc, View};
 use fraction::{Fraction, One};
 use futures::stream::{self, StreamExt};
 use std::collections::HashSet;
 use std::time::Duration;
-use tests::{MixNode, Node, NomosNode, SpawnConfig};
+use tests::{ConsensusConfig, MixNode, Node, NomosNode, SpawnConfig};
 
 const TARGET_VIEW: View = View::new(20);
 
+#[derive(serde::Serialize)]
+struct Info {
+    node_id: String,
+    block_id: String,
+    view: View,
+}
+
 async fn happy_test(nodes: Vec<NomosNode>) {
     let timeout = std::time::Duration::from_secs(20);
     let timeout = tokio::time::sleep(timeout);
@@ -43,18 +50,40 @@ async fn happy_test(nodes: Vec<NomosNode>) {
                 .unwrap()
         })
         .collect::<HashSet<_>>();
+
+    // try to see if we have invalid blocks
+    let invalid_blocks = infos
+        .iter()
+        .flat_map(|i| {
+            i.safe_blocks.values().filter_map(|b| match &b.parent_qc {
+                Qc::Standard(_) => None,
+                Qc::Aggregated(_) => Some(Info {
+                    node_id: i.id.to_string(),
+                    block_id: b.id.to_string(),
+                    view: b.view,
+                }),
+            })
+        })
+        .collect::<Vec<_>>();
+
+    assert!(
+        invalid_blocks.is_empty(),
+        "{}",
+        serde_json::to_string_pretty(&invalid_blocks).unwrap()
+    );
     assert_eq!(blocks.len(), 1);
 }
 
 #[tokio::test]
 async fn two_nodes_happy() {
-    let (_mixnodes, mixnet_node_configs, mixnet_topology) = MixNode::spawn_nodes(2).await;
+    let (_mixnodes, mixnet_config) = MixNode::spawn_nodes(2).await;
-    let nodes = NomosNode::spawn_nodes(SpawnConfig::Star {
+    let nodes = NomosNode::spawn_nodes(SpawnConfig::Chain {
+        consensus: ConsensusConfig {
             n_participants: 2,
             threshold: Fraction::one(),
             timeout: Duration::from_secs(10),
-        mixnet_node_configs,
+        },
-        mixnet_topology,
+        mixnet: mixnet_config,
     })
     .await;
     happy_test(nodes).await;
@@ -62,13 +91,14 @@ async fn two_nodes_happy() {
 
 #[tokio::test]
 async fn ten_nodes_happy() {
-    let (_mixnodes, mixnet_node_configs, mixnet_topology) = MixNode::spawn_nodes(3).await;
+    let (_mixnodes, mixnet_config) = MixNode::spawn_nodes(3).await;
-    let nodes = NomosNode::spawn_nodes(SpawnConfig::Star {
+    let nodes = NomosNode::spawn_nodes(SpawnConfig::Chain {
+        consensus: ConsensusConfig {
             n_participants: 10,
             threshold: Fraction::one(),
             timeout: Duration::from_secs(10),
-        mixnet_node_configs,
+        },
-        mixnet_topology,
+        mixnet: mixnet_config,
     })
     .await;
     happy_test(nodes).await;

tests/src/tests/unhappy.rs

@@ -2,19 +2,20 @@ use consensus_engine::View;
 use fraction::Fraction;
 use futures::stream::{self, StreamExt};
 use std::collections::HashSet;
-use tests::{MixNode, Node, NomosNode, SpawnConfig};
+use tests::{ConsensusConfig, MixNode, Node, NomosNode, SpawnConfig};
 
 const TARGET_VIEW: View = View::new(20);
 
 #[tokio::test]
 async fn ten_nodes_one_down() {
-    let (_mixnodes, mixnet_node_configs, mixnet_topology) = MixNode::spawn_nodes(3).await;
+    let (_mixnodes, mixnet_config) = MixNode::spawn_nodes(3).await;
-    let mut nodes = NomosNode::spawn_nodes(SpawnConfig::Star {
+    let mut nodes = NomosNode::spawn_nodes(SpawnConfig::Chain {
+        consensus: ConsensusConfig {
             n_participants: 10,
             threshold: Fraction::new(9u32, 10u32),
             timeout: std::time::Duration::from_secs(5),
-        mixnet_node_configs,
+        },
-        mixnet_topology,
+        mixnet: mixnet_config,
     })
     .await;
     let mut failed_node = nodes.pop().unwrap();