Kickstart event building in sim app

nomos-services/consensus/src/lib.rs

@@ -542,7 +542,7 @@ where
     }
 }
 
-enum Event<Tx: Clone + Hash + Eq> {
+pub enum Event<Tx: Clone + Hash + Eq> {
     Proposal {
         block: Block<Tx>,
         stream: Pin<Box<dyn Stream<Item = Block<Tx>> + Send>>,

simulations/Cargo.toml

@@ -12,7 +12,9 @@ clap = { version = "4", features = ["derive"] }
 crc32fast = "1.3"
 crossbeam = { version = "0.8.2", features = ["crossbeam-channel"] }
 fixed-slice-deque = "0.1.0-beta2"
+futures = "0.3"
 nomos-core = { path = "../nomos-core" }
+nomos-consensus = { path = "../nomos-services/consensus" }
 polars = { version = "0.27", features = ["serde", "object", "json", "csv-file", "parquet", "dtype-struct"] }
 rand = { version = "0.8", features = ["small_rng"] }
 rayon = "1.7"

simulations/src/node/carnot/event_builder.rs

@@ -0,0 +1,63 @@
+use crate::node::carnot::messages::CarnotMessage;
+use nomos_consensus::network::messages::VoteMsg;
+use nomos_consensus::Event;
+use nomos_consensus::Event::TimeoutQc;
+use nomos_core::block::Block;
+use std::collections::HashMap;
+
+pub type CarnotTx = [u8; 32];
+
+#[derive(Default)]
+pub struct EventBuilderConfiguration {
+    pub votes_threshold: usize,
+}
+
+pub struct EventBuilder {
+    vote_message: HashMap<View, Vec<VoteMsg>>,
+    config: EventBuilderConfiguration,
+}
+
+impl EventBuilder {
+    pub fn new() -> Self {
+        Self {
+            vote_message: HashMap::new(),
+            config: Default::default(),
+        }
+    }
+
+    pub fn step(&mut self, messages: &[CarnotMessage]) -> Vec<Event<CarnotTx>> {
+        let mut events = Vec::new();
+        for message in messages {
+            match message {
+                CarnotMessage::Proposal(msg) => {
+                    events.push(Event::Proposal {
+                        block: Block::from_bytes(&msg.chunk),
+                        stream: Box::pin(futures::stream::empty()),
+                    });
+                }
+                CarnotMessage::TimeoutQc(msg) => {
+                    events.push(Event::TimeoutQc {
+                        timeout_qc: msg.qc.clone(),
+                    });
+                }
+                CarnotMessage::Vote(msg) => {
+                    let msg_view = msg.vote.view;
+                    let entry = self.vote_message.entry(msg_view).or_default();
+                    entry.push(msg.clone());
+                    if entry.len() >= self.config.votes_threshold {
+                        events.push(Event::Approve {
+                            qc: msg.qc.cloned().unwrap(),
+                            block: msg.vote.block,
+                            votes: entry.iter().cloned().collect(),
+                        })
+                    }
+                    self.vote_message.remove(&msg_view);
+                }
+                CarnotMessage::Timeout(_) => {}
+                CarnotMessage::NewView(_) => {}
+            }
+        }
+
+        events
+    }
+}

simulations/src/node/carnot/messages.rs

@@ -0,0 +1,9 @@
+use nomos_consensus::network::messages::{NewViewMsg, ProposalChunkMsg, TimeoutQcMsg, VoteMsg};
+
+pub(crate) enum CarnotMessage {
+    Proposal(ProposalChunkMsg),
+    Vote(VoteMsg),
+    TimeoutQc(TimeoutQcMsg),
+    Timeout(TimeoutQcMsg),
+    NewView(NewViewMsg),
+}

simulations/src/node/carnot/mod.rs

@@ -1,3 +1,6 @@
+mod event_builder;
+mod messages;
+
 // std
 // crates
 use serde::{Deserialize, Serialize};