Linear leios tests

sim-rs/CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## Unreleased
+
+### Linear Leios
+- Add some protocol-level tests
+- Fix bug; transactions with conflicts referenced by EBs did not propagate far enough
+
 ## v1.3.0
 
 ### Linear Leios

sim-rs/sim-core/Cargo.toml

@@ -7,6 +7,7 @@ rust-version = "1.88"
 [dependencies]
 anyhow = "1"
 async-stream = "0.3"
+dashmap = "6"
 futures = "0.3"
 netsim-async = { git = "https://github.com/input-output-hk/ce-netsim.git", rev = "9d1e26c" }
 num-traits = "0.2"
@@ -20,4 +21,5 @@ tokio-util = "0.7"
 tracing = "0.1"
 
 [dev-dependencies]
+serde_yaml = "0.9"
 tokio = { version = "1", features = ["macros", "rt"] }

sim-rs/sim-core/src/clock.rs

@@ -10,11 +10,13 @@ use std::{
 pub use coordinator::ClockCoordinator;
 use coordinator::ClockEvent;
 use futures::FutureExt;
+pub use mock::MockClockCoordinator;
 use timestamp::AtomicTimestamp;
 pub use timestamp::Timestamp;
 use tokio::sync::{mpsc, oneshot};
 
 mod coordinator;
+mod mock;
 mod timestamp;
 
 // wrapper struct which holds a SimulationEvent,

sim-rs/sim-core/src/clock/mock.rs

@@ -0,0 +1,106 @@
+use std::{
+    collections::HashMap,
+    sync::{Arc, atomic::AtomicUsize},
+    time::Duration,
+};
+
+use tokio::sync::{mpsc, oneshot};
+
+use crate::clock::{
+    Clock, TaskInitiator, Timestamp, coordinator::ClockEvent, timestamp::AtomicTimestamp,
+};
+
+pub struct MockClockCoordinator {
+    time: Arc<AtomicTimestamp>,
+    tx: mpsc::UnboundedSender<ClockEvent>,
+    rx: mpsc::UnboundedReceiver<ClockEvent>,
+    waiter_count: Arc<AtomicUsize>,
+    tasks: Arc<AtomicUsize>,
+    waiters: HashMap<usize, Waiter>,
+}
+
+impl Default for MockClockCoordinator {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl MockClockCoordinator {
+    pub fn new() -> Self {
+        let time = Arc::new(AtomicTimestamp::new(Timestamp::zero()));
+        let (tx, rx) = mpsc::unbounded_channel();
+        let waiter_count = Arc::new(AtomicUsize::new(0));
+        let tasks = Arc::new(AtomicUsize::new(0));
+        Self {
+            time,
+            tx,
+            rx,
+            waiter_count,
+            tasks,
+            waiters: HashMap::new(),
+        }
+    }
+
+    pub fn clock(&self) -> Clock {
+        Clock::new(
+            Duration::from_nanos(1),
+            self.time.clone(),
+            self.waiter_count.clone(),
+            TaskInitiator::new(self.tasks.clone()),
+            self.tx.clone(),
+        )
+    }
+
+    pub fn now(&self) -> Timestamp {
+        self.time.load(std::sync::atomic::Ordering::Acquire)
+    }
+
+    pub fn advance_time(&mut self, until: Timestamp) {
+        while let Ok(event) = self.rx.try_recv() {
+            match event {
+                ClockEvent::Wait { actor, until, done } => {
+                    if self.waiters.insert(actor, Waiter { until, done }).is_some() {
+                        panic!("waiter {actor} waited twice");
+                    }
+                }
+                ClockEvent::CancelWait { actor } => {
+                    if self.waiters.remove(&actor).is_none() {
+                        panic!("waiter {actor} cancelled a wait twice");
+                    }
+                }
+                ClockEvent::FinishTask => {
+                    if self.tasks.fetch_sub(1, std::sync::atomic::Ordering::AcqRel) == 0 {
+                        panic!("cancelled too many tasks");
+                    }
+                }
+            }
+        }
+        assert_eq!(
+            self.waiters.len(),
+            self.waiter_count.load(std::sync::atomic::Ordering::Acquire),
+            "not every worker is waiting for time to pass"
+        );
+
+        self.time.store(until, std::sync::atomic::Ordering::Release);
+        self.waiters = std::mem::take(&mut self.waiters)
+            .into_iter()
+            .filter_map(|(actor, waiter)| {
+                if let Some(t) = &waiter.until {
+                    if *t < until {
+                        panic!("advanced time too far (waited for {until:?}, next event at {t:?})");
+                    }
+                    if *t == until {
+                        let _ = waiter.done.send(());
+                        return None;
+                    }
+                }
+                Some((actor, waiter))
+            })
+            .collect();
+    }
+}
+
+struct Waiter {
+    until: Option<Timestamp>,
+    done: oneshot::Sender<()>,
+}

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

@@ -84,7 +84,7 @@ impl Block {
     }
 }
 
-#[derive(Clone, Debug)]
+#[derive(Clone, Debug, PartialEq, Eq)]
 pub struct LinearRankingBlockHeader {
     pub id: BlockId,
     pub vrf: u64,
@@ -93,7 +93,7 @@ pub struct LinearRankingBlockHeader {
     pub eb_announcement: Option<EndorserBlockId>,
 }
 
-#[derive(Clone, Debug)]
+#[derive(Clone, Debug, PartialEq, Eq)]
 pub struct LinearRankingBlock {
     pub header: LinearRankingBlockHeader,
     pub transactions: Vec<Arc<Transaction>>,
@@ -227,7 +227,7 @@ impl StracciatellaEndorserBlock {
     }
 }
 
-#[derive(Debug)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub struct LinearEndorserBlock {
     pub slot: u64,
     pub producer: NodeId,
@@ -293,7 +293,7 @@ pub enum TransactionLostReason {
     EBExpired,
 }
 
-#[derive(Clone, Debug, Serialize)]
+#[derive(Clone, Debug, Serialize, PartialEq, Eq)]
 pub struct Endorsement<Node: Display = NodeId> {
     pub eb: EndorserBlockId<Node>,
     pub size_bytes: u64,

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

@@ -28,6 +28,8 @@ mod linear_leios;
 mod lottery;
 mod slot;
 mod stracciatella;
+#[cfg(test)]
+mod tests;
 mod tx;
 
 enum NetworkWrapper {
@@ -304,6 +306,13 @@ impl<N: NodeImpl> Default for EventResult<N> {
 }
 
 impl<N: NodeImpl> EventResult<N> {
+    #[cfg(test)]
+    pub fn merge(&mut self, mut other: EventResult<N>) {
+        self.messages.append(&mut other.messages);
+        self.tasks.append(&mut other.tasks);
+        self.timed_events.append(&mut other.timed_events);
+    }
+
     pub fn send_to(&mut self, to: NodeId, msg: N::Message) {
         self.messages.push((to, msg));
     }