sim-rs: emit TXLost events when transactions are lost

Author: SupernaviX

sim-rs/sim-cli/src/events.rs

@@ -5,6 +5,7 @@ use anyhow::Result;
 use async_compression::tokio::write::GzipEncoder;
 use average::Variance;
 use itertools::Itertools as _;
+use liveness::LivenessMonitor;
 use pretty_bytes_rust::{pretty_bytes, PrettyBytesOptions};
 use serde::Serialize;
 use sim_core::{
@@ -21,6 +22,7 @@ use tokio::{
 use tracing::{info, info_span};
 
 mod aggregate;
+mod liveness;
 
 type InputBlockId = sim_core::model::InputBlockId<Node>;
 type EndorserBlockId = sim_core::model::EndorserBlockId<Node>;
@@ -46,7 +48,7 @@ pub struct EventMonitor {
     pool_ids: Vec<NodeId>,
     maximum_ib_age: u64,
     maximum_eb_age: u64,
-    events_source: mpsc::UnboundedReceiver<(Event, Timestamp)>,
+    events_source: LivenessMonitor,
     output_path: Option<PathBuf>,
     aggregate: bool,
 }
@@ -71,7 +73,7 @@ impl EventMonitor {
             pool_ids,
             maximum_ib_age,
             maximum_eb_age: config.max_eb_age,
-            events_source,
+            events_source: LivenessMonitor::new(config, events_source),
             output_path,
             aggregate: config.aggregate_events,
         }
@@ -185,6 +187,7 @@ impl EventMonitor {
                 Event::TXReceived { .. } => {
                     tx_messages.received += 1;
                 }
+                Event::TXLost { .. } => {}
                 Event::RBLotteryWon { .. } => {}
                 Event::RBGenerated {
                     id: BlockId { slot, producer },

sim-rs/sim-cli/src/events/liveness.rs

@@ -0,0 +1,231 @@
+use std::collections::{BTreeMap, BTreeSet, VecDeque};
+
+use sim_core::{
+    clock::Timestamp,
+    config::SimConfiguration,
+    events::{BlockRef, Event},
+    model::{TransactionId, TransactionLostReason},
+};
+use tokio::sync::mpsc;
+
+use super::{EndorserBlockId, InputBlockId};
+
+/// Emits additional events when it is no longer possible for a transaction to reach the chain.
+pub struct LivenessMonitor {
+    events_source: mpsc::UnboundedReceiver<(Event, Timestamp)>,
+    queue: VecDeque<(Event, Timestamp)>,
+    txs: BTreeMap<TransactionId, MonitoredTX>,
+    ibs: BTreeMap<InputBlockId, MonitoredIB>,
+    ebs: BTreeMap<EndorserBlockId, MonitoredEB>,
+    stage_length: u64,
+    vote_threshold: u64,
+}
+
+impl LivenessMonitor {
+    pub fn new(
+        config: &SimConfiguration,
+        events_source: mpsc::UnboundedReceiver<(Event, Timestamp)>,
+    ) -> Self {
+        Self {
+            events_source,
+            queue: VecDeque::new(),
+            txs: BTreeMap::new(),
+            ibs: BTreeMap::new(),
+            ebs: BTreeMap::new(),
+            stage_length: config.stage_length,
+            vote_threshold: config.vote_threshold,
+        }
+    }
+
+    pub async fn recv(&mut self) -> Option<(Event, Timestamp)> {
+        if let Some(next) = self.queue.pop_front() {
+            return Some(next);
+        }
+        let (event, time) = self.events_source.recv().await?;
+        match &event {
+            Event::TXGenerated { id, .. } => {
+                self.txs.insert(*id, MonitoredTX::new());
+            }
+            Event::IBGenerated {
+                id, transactions, ..
+            } => {
+                for tx_id in transactions {
+                    if let Some(tx) = self.txs.get_mut(tx_id) {
+                        tx.ibs.insert(id.clone());
+                        tx.last_ib_pipeline = std::cmp::max(tx.last_ib_pipeline, Some(id.pipeline));
+                    }
+                }
+                self.ibs
+                    .insert(id.clone(), MonitoredIB::new(transactions.clone()));
+            }
+            Event::EBGenerated {
+                id,
+                transactions,
+                input_blocks,
+                endorser_blocks,
+                ..
+            } => {
+                for BlockRef { id: tx_id } in transactions {
+                    if let Some(tx) = self.txs.get_mut(tx_id) {
+                        tx.ebs.insert(id.clone());
+                        tx.last_eb_pipeline = std::cmp::max(tx.last_eb_pipeline, Some(id.pipeline));
+                    }
+                }
+                for BlockRef { id: ib_id } in input_blocks {
+                    if let Some(ib) = self.ibs.get(ib_id) {
+                        for tx_id in &ib.txs {
+                            if let Some(tx) = self.txs.get_mut(tx_id) {
+                                tx.ebs.insert(id.clone());
+                                tx.last_eb_pipeline =
+                                    std::cmp::max(tx.last_eb_pipeline, Some(id.pipeline));
+                            }
+                        }
+                    }
+                }
+                self.ebs.insert(
+                    id.clone(),
+                    MonitoredEB::new(
+                        transactions.iter().map(|tx| tx.id).collect(),
+                        input_blocks.iter().map(|ib| ib.id.clone()).collect(),
+                        endorser_blocks.iter().map(|eb| eb.id.clone()).collect(),
+                    ),
+                );
+            }
+            Event::VTBundleGenerated { votes, .. } => {
+                for (eb_id, count) in &votes.0 {
+                    let eb = self.ebs.get_mut(eb_id).unwrap();
+                    eb.votes += *count as u64;
+                    if eb.votes >= self.vote_threshold {
+                        for ib_id in &eb.ibs {
+                            if let Some(ib) = self.ibs.get_mut(ib_id) {
+                                for tx_id in &ib.txs {
+                                    if let Some(tx) = self.txs.get_mut(tx_id) {
+                                        tx.certified_ebs.insert(eb_id.clone());
+                                    }
+                                }
+                            }
+                        }
+                        for tx_id in &eb.txs {
+                            if let Some(tx) = self.txs.get_mut(tx_id) {
+                                tx.certified_ebs.insert(eb_id.clone());
+                            }
+                        }
+                    }
+                }
+            }
+            Event::RBGenerated {
+                endorsement,
+                transactions,
+                ..
+            } => {
+                for tx in transactions {
+                    self.txs.remove(tx);
+                }
+                if let Some(endorsement) = endorsement {
+                    let mut eb_queue = vec![endorsement.eb.id.clone()];
+                    while let Some(eb_id) = eb_queue.pop() {
+                        let Some(eb) = self.ebs.remove(&eb_id) else {
+                            continue;
+                        };
+                        self.txs.retain(|_, tx| !tx.ebs.contains(&eb_id));
+                        for ib_id in &eb.ibs {
+                            self.ibs.remove(ib_id);
+                        }
+                        eb_queue.extend(eb.ebs);
+                    }
+                }
+            }
+            Event::GlobalSlot { slot } => {
+                if slot % self.stage_length == 0 {
+                    let pipeline = slot / self.stage_length;
+                    self.handle_new_pipeline(pipeline, time);
+                }
+            }
+            _ => {}
+        }
+        Some((event, time))
+    }
+
+    fn handle_new_pipeline(&mut self, pipeline: u64, time: Timestamp) {
+        self.txs.retain(|id, tx| {
+            if tx.ebs.is_empty()
+                && tx
+                    .last_ib_pipeline
+                    .is_some_and(|ib_pipeline| ib_pipeline + 4 < pipeline)
+            {
+                // this transaction was only in IBs which never reached any EBs.
+                self.queue.push_back((
+                    Event::TXLost {
+                        id: *id,
+                        reason: TransactionLostReason::IBExpired,
+                    },
+                    time,
+                ));
+                return false;
+            }
+
+            if tx.certified_ebs.is_empty()
+                && tx
+                    .last_eb_pipeline
+                    .is_some_and(|eb_pipeline| eb_pipeline + 1 < pipeline)
+            {
+                // this transaction was only in EBs which were never certified
+                self.queue.push_back((
+                    Event::TXLost {
+                        id: *id,
+                        reason: TransactionLostReason::EBExpired,
+                    },
+                    time,
+                ));
+                return false;
+            }
+
+            true
+        });
+    }
+}
+
+struct MonitoredTX {
+    last_ib_pipeline: Option<u64>,
+    ibs: BTreeSet<InputBlockId>,
+    last_eb_pipeline: Option<u64>,
+    ebs: BTreeSet<EndorserBlockId>,
+    certified_ebs: BTreeSet<EndorserBlockId>,
+}
+impl MonitoredTX {
+    fn new() -> Self {
+        Self {
+            last_ib_pipeline: None,
+            ibs: BTreeSet::new(),
+            last_eb_pipeline: None,
+            ebs: BTreeSet::new(),
+            certified_ebs: BTreeSet::new(),
+        }
+    }
+}
+
+struct MonitoredIB {
+    txs: Vec<TransactionId>,
+}
+impl MonitoredIB {
+    fn new(txs: Vec<TransactionId>) -> Self {
+        Self { txs }
+    }
+}
+
+struct MonitoredEB {
+    txs: Vec<TransactionId>,
+    ibs: Vec<InputBlockId>,
+    ebs: Vec<EndorserBlockId>,
+    votes: u64,
+}
+impl MonitoredEB {
+    fn new(txs: Vec<TransactionId>, ibs: Vec<InputBlockId>, ebs: Vec<EndorserBlockId>) -> Self {
+        Self {
+            txs,
+            ibs,
+            ebs,
+            votes: 0,
+        }
+    }
+}

sim-rs/sim-core/src/config.rs

@@ -438,6 +438,7 @@ pub struct SimConfiguration {
     pub max_eb_age: u64,
     pub late_ib_inclusion: bool,
     pub variant: LeiosVariant,
+    pub vote_threshold: u64,
     pub(crate) praos_fallback: bool,
     pub(crate) header_diffusion_time: Duration,
     pub(crate) relay_strategy: RelayStrategy,
@@ -446,7 +447,6 @@ pub struct SimConfiguration {
     pub(crate) ib_generation_probability: f64,
     pub(crate) eb_generation_probability: f64,
     pub(crate) vote_probability: f64,
-    pub(crate) vote_threshold: u64,
     pub(crate) vote_slot_length: u64,
     pub(crate) max_block_size: u64,
     pub(crate) max_ib_size: u64,

sim-rs/sim-core/src/events.rs

@@ -9,7 +9,7 @@ use crate::{
     config::{NodeConfiguration, NodeId},
     model::{
         Block, BlockId, CpuTaskId, EndorserBlockId, InputBlockId, NoVoteReason, Transaction,
-        TransactionId, VoteBundle, VoteBundleId,
+        TransactionId, TransactionLostReason, VoteBundle, VoteBundleId,
     },
 };
 
@@ -115,6 +115,10 @@ pub enum Event {
         sender: Node,
         recipient: Node,
     },
+    TXLost {
+        id: TransactionId,
+        reason: TransactionLostReason,
+    },
     RBLotteryWon {
         id: BlockId<Node>,
         slot: u64,

sim-rs/sim-core/src/model.rs

@@ -218,6 +218,12 @@ pub enum NoVoteReason {
     UncertifiedEBReference,
 }
 
+#[derive(Debug, Clone, Serialize)]
+pub enum TransactionLostReason {
+    IBExpired,
+    EBExpired,
+}
+
 #[derive(Clone, Debug, Serialize)]
 pub struct Endorsement<Node: Display = NodeId> {
     pub eb: EndorserBlockId<Node>,

sim-rs/sim-core/src/sim/node.rs

@@ -1336,7 +1336,7 @@ impl Node {
                 .entry(votes.id.producer)
                 .or_default();
             *eb_votes += count;
-            if *eb_votes as u64 > self.sim_config.vote_threshold {
+            if *eb_votes as u64 >= self.sim_config.vote_threshold {
                 self.leios
                     .earliest_eb_cert_times_by_pipeline
                     .entry(eb.pipeline)