sim-rs: fix bug with conflicting TXs

Author: SupernaviX

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/src/clock/mock.rs

@@ -51,6 +51,10 @@ impl MockClockCoordinator {
         )
     }
 
+    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 {

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

@@ -1004,7 +1004,7 @@ impl LinearLeiosNode {
 
         if missing_txs.is_empty() {
             self.queued
-                .schedule_cpu_task(CpuTask::EBBlockValidated(eb, seen));
+                .schedule_cpu_task(CpuTask::EBBlockValidated(eb.clone(), seen));
         } else {
             for tx_id in missing_txs {
                 self.leios
@@ -1014,6 +1014,23 @@ impl LinearLeiosNode {
                     .push(eb.id());
             }
         }
+
+        if matches!(
+            self.sim_config.variant,
+            LeiosVariant::LinearWithTxReferences
+        ) {
+            // If the EB references any TXs which we already have, but are not in our mempool,
+            // we must have failed to add them to the mempool due to conflicts.
+            // Announce those TXs to our peers, since we didn't before.
+            for tx in &eb.txs {
+                if !self.has_tx(tx.id) || self.praos.mempool.contains_key(&tx.id) {
+                    continue;
+                }
+                for peer in &self.consumers {
+                    self.queued.send_to(*peer, Message::AnnounceTx(tx.id));
+                }
+            }
+        }
     }
 
     fn try_validating_eb(&mut self, eb_id: EndorserBlockId) {

sim-rs/sim-core/src/sim/tests/linear_leios.rs

@@ -17,9 +17,16 @@ use crate::{
 };
 
 fn new_sim_config(topology: RawTopology) -> Arc<SimConfiguration> {
-    let params =
+    let mut params: crate::config::RawParameters =
         serde_yaml::from_slice(include_bytes!("../../../../parameters/config.default.yaml"))
             .unwrap();
+    params.leios_variant = crate::config::LeiosVariant::LinearWithTxReferences;
+    // every transaction fills up exactly half of an RB
+    let tx_size = params.rb_body_max_size_bytes / 2;
+    params.tx_size_bytes_distribution = crate::config::DistributionConfig::Constant {
+        value: tx_size as f64,
+    };
+    params.tx_max_size_bytes = tx_size;
     let topology = topology.into();
     Arc::new(SimConfiguration::build(params, topology).unwrap())
 }
@@ -154,6 +161,49 @@ impl TestDriver {
         }
     }
 
+    pub fn expect_tx_sent(&mut self, from: NodeId, to: NodeId, tx: Arc<Transaction>) {
+        self.expect_message(from, to, Message::AnnounceTx(tx.id));
+        self.expect_message(to, from, Message::RequestTx(tx.id));
+        self.expect_message(from, to, Message::Tx(tx.clone()));
+        self.expect_cpu_task(to, CpuTask::TransactionValidated(from, tx));
+    }
+
+    pub fn expect_tx_not_sent(&mut self, from: NodeId, to: NodeId, tx: Arc<Transaction>) {
+        self.expect_no_message(from, to, Message::AnnounceTx(tx.id));
+    }
+
+    pub fn expect_rb_and_eb_sent(
+        &mut self,
+        from: NodeId,
+        to: NodeId,
+        rb: Arc<LinearRankingBlock>,
+        eb: Option<Arc<LinearEndorserBlock>>,
+    ) {
+        self.expect_message(from, to, Message::AnnounceRBHeader(rb.header.id));
+        self.expect_message(to, from, Message::RequestRBHeader(rb.header.id));
+        self.expect_message(
+            from,
+            to,
+            Message::RBHeader(rb.header.clone(), true, eb.is_some()),
+        );
+        self.expect_cpu_task(
+            to,
+            CpuTask::RBHeaderValidated(from, rb.header.clone(), true, eb.is_some()),
+        );
+        self.expect_message(to, from, Message::RequestRB(rb.header.id));
+        self.expect_message(from, to, Message::RB(rb.clone()));
+        self.expect_cpu_task(to, CpuTask::RBBlockValidated(rb));
+        if let Some(eb) = eb {
+            self.expect_message(to, from, Message::RequestEB(eb.id()));
+            self.expect_message(from, to, Message::EB(eb.clone()));
+            self.expect_cpu_task(to, CpuTask::EBHeaderValidated(from, eb));
+        }
+    }
+
+    pub fn expect_eb_validated(&mut self, node: NodeId, eb: Arc<LinearEndorserBlock>) {
+        self.expect_cpu_task(node, CpuTask::EBBlockValidated(eb, self.time.now()));
+    }
+
     pub fn expect_message(
         &mut self,
         from: NodeId,
@@ -172,7 +222,12 @@ impl TestDriver {
         });
         assert!(
             found,
-            "message {message:?} was not sent from {from} to {to}"
+            "message {message:?} was not sent from {from} to {to}\npending messages: {:?}",
+            queued
+                .messages
+                .iter()
+                .filter(|(t, _)| t == &to)
+                .collect::<Vec<_>>(),
         );
         let events = self
             .nodes
@@ -182,6 +237,20 @@ impl TestDriver {
         self.queued.entry(to).or_default().merge(events);
     }
 
+    pub fn expect_no_message(
+        &mut self,
+        from: NodeId,
+        to: NodeId,
+        message: <LinearLeiosNode as NodeImpl>::Message,
+    ) {
+        let Some(queued) = self.queued.get(&from) else {
+            return;
+        };
+        for (t, m) in &queued.messages {
+            assert_ne!((t, m), (&to, &message));
+        }
+    }
+
     pub fn expect_cpu_task(&mut self, node: NodeId, task: <LinearLeiosNode as NodeImpl>::Task) {
         self.expect_cpu_task_matching(node, |t| if *t == task { Some(t.clone()) } else { None });
     }
@@ -212,11 +281,14 @@ impl TestDriver {
     }
 }
 
-fn is_new_rb_task(task: &CpuTask) -> Option<(LinearRankingBlock, Option<LinearEndorserBlock>)> {
+fn is_new_rb_task(
+    task: &CpuTask,
+) -> Option<(Arc<LinearRankingBlock>, Option<Arc<LinearEndorserBlock>>)> {
     match task {
-        CpuTask::RBBlockGenerated(rb, eb) => {
-            Some((rb.clone(), eb.as_ref().map(|(eb, _)| eb.clone())))
-        }
+        CpuTask::RBBlockGenerated(rb, eb) => Some((
+            Arc::new(rb.clone()),
+            eb.as_ref().map(|(eb, _)| Arc::new(eb.clone())),
+        )),
         _ => None,
     }
 }
@@ -233,17 +305,11 @@ fn should_propagate_transactions() {
 
     // Node 1 produces a transaction, node 2 should request it
     let tx1 = sim.produce_tx(node1, false);
-    sim.expect_message(node1, node2, Message::AnnounceTx(tx1.id));
-    sim.expect_message(node2, node1, Message::RequestTx(tx1.id));
-    sim.expect_message(node1, node2, Message::Tx(tx1.clone()));
-    sim.expect_cpu_task(node2, CpuTask::TransactionValidated(node1, tx1.clone()));
+    sim.expect_tx_sent(node1, node2, tx1.clone());
 
     // Node 2 produces a transaction, node 1 should request it
     let tx2 = sim.produce_tx(node2, false);
-    sim.expect_message(node2, node1, Message::AnnounceTx(tx2.id));
-    sim.expect_message(node1, node2, Message::RequestTx(tx2.id));
-    sim.expect_message(node2, node1, Message::Tx(tx2.clone()));
-    sim.expect_cpu_task(node1, CpuTask::TransactionValidated(node2, tx2.clone()));
+    sim.expect_tx_sent(node2, node1, tx2.clone());
 
     // When node 1 produces an RB, it should include both TXs
     sim.win_next_rb_lottery(node1, 0);
@@ -252,3 +318,82 @@ fn should_propagate_transactions() {
     assert_eq!(new_rb.transactions, vec![tx1, tx2]);
     assert_eq!(new_eb, None);
 }
+
+#[test]
+fn should_not_propagate_conflicting_transactions() {
+    let topology = new_topology(vec![
+        ("node-1", new_node(Some(1000), vec!["node-2"])),
+        ("node-2", new_node(Some(1000), vec!["node-1"])),
+    ]);
+    let mut sim = TestDriver::new(topology);
+    let node1 = sim.id_for("node-1");
+    let node2 = sim.id_for("node-2");
+
+    // Node 1 and 2 produce conflicting transactions
+    let tx1 = sim.produce_tx(node1, false);
+    let tx2 = sim.produce_tx(node2, true);
+
+    // Each node should send its TX to the other node,
+    sim.expect_tx_sent(node1, node2, tx1.clone());
+    sim.expect_tx_sent(node2, node1, tx2.clone());
+
+    // When node 1 produces an RB, it should include only its own TX
+    sim.win_next_rb_lottery(node1, 0);
+    sim.next_slot();
+    let (new_rb, new_eb) = sim.expect_cpu_task_matching(node1, is_new_rb_task);
+    assert_eq!(new_rb.transactions, vec![tx1]);
+    assert_eq!(new_eb, None);
+}
+
+#[test]
+fn should_repropagate_conflicting_transactions_from_eb() {
+    let topology = new_topology(vec![
+        ("node-1", new_node(Some(1000), vec!["node-2"])),
+        ("node-2", new_node(Some(1000), vec!["node-1", "node-3"])),
+        ("node-3", new_node(Some(1000), vec!["node-2"])),
+    ]);
+    let mut sim = TestDriver::new(topology);
+    let node1 = sim.id_for("node-1");
+    let node2 = sim.id_for("node-2");
+    let node3 = sim.id_for("node-3");
+
+    // Node 1 produces 3 transactions
+    let tx1_1 = sim.produce_tx(node1, false);
+    let tx1_2 = sim.produce_tx(node1, false);
+    let tx1_3 = sim.produce_tx(node1, false);
+
+    // Node 2 produces a transaction which conflicts with Node 1's final transaction
+    let tx2 = sim.produce_tx(node2, true);
+    // Node 2 sends its transactions to nodes 1 and 3
+    sim.expect_tx_sent(node2, node1, tx2.clone());
+    sim.expect_tx_sent(node2, node3, tx2.clone());
+
+    // Node 1 sends all of its transactions to node 2
+    sim.expect_tx_sent(node1, node2, tx1_1.clone());
+    sim.expect_tx_sent(node1, node2, tx1_2.clone());
+    sim.expect_tx_sent(node1, node2, tx1_3.clone());
+
+    // Node 2 sends the first two transactions to node 3, but not the conflicting third
+    sim.expect_tx_sent(node2, node3, tx1_1.clone());
+    sim.expect_tx_sent(node2, node3, tx1_2.clone());
+    sim.expect_tx_not_sent(node2, node3, tx1_3.clone());
+
+    // Now, Node 1 produces an RB (with an EB, because there are enough transactions)
+    sim.win_next_rb_lottery(node1, 0);
+    sim.next_slot();
+    let (rb, eb) = sim.expect_cpu_task_matching(node1, is_new_rb_task);
+    let eb = eb.expect("node did not produce EB");
+    assert_eq!(rb.transactions, vec![tx1_1, tx1_2]);
+    assert_eq!(eb.txs, vec![tx1_3.clone()]);
+
+    // That RB and EB propagate from node 1 to node 2
+    sim.expect_rb_and_eb_sent(node1, node2, rb.clone(), Some(eb.clone()));
+    // Node 2 fully validates the EB, because node 1 has all TXs
+    sim.expect_eb_validated(node2, eb.clone());
+    // And Node 2 propagates it to Node 3
+    sim.expect_rb_and_eb_sent(node2, node3, rb.clone(), Some(eb.clone()));
+
+    // and NOW Node 2 will tell Node 3 about the EB's conflicting TX
+    sim.expect_tx_sent(node2, node3, tx1_3);
+    sim.expect_eb_validated(node3, eb);
+}