Merge pull request #308 from carlaKC/determinism-fixups

multi: Small Determinism Fixups

Author: carlaKC

sim-cli/src/main.rs

@@ -39,7 +39,10 @@ async fn main() -> anyhow::Result<()> {
     } else {
         let latency = cli.latency_ms.unwrap_or(0);
         let interceptors = if latency > 0 {
-            vec![Arc::new(LatencyIntercepor::new_poisson(latency as f32)?) as Arc<dyn Interceptor>]
+            vec![Arc::new(LatencyIntercepor::new_poisson(
+                latency as f32,
+                cli.fix_seed,
+            )?) as Arc<dyn Interceptor>]
         } else {
             vec![]
         };

simln-lib/src/latency_interceptor.rs

@@ -3,7 +3,9 @@ use crate::sim_node::{
 };
 use crate::SimulationError;
 use async_trait::async_trait;
+use rand::{rngs::StdRng, SeedableRng};
 use rand_distr::{Distribution, Poisson};
+use std::sync::Mutex;
 use std::time::Duration;
 use tokio::{select, time};
 
@@ -13,20 +15,34 @@ where
     D: Distribution<f32> + Send + Sync,
 {
     latency_dist: D,
+    /// Seedable RNG used to sample the latency distribution. Held behind a mutex because the interceptor is shared
+    /// across concurrent HTLCs, and seeded (rather than `thread_rng`) so that simulation runs are reproducible.
+    rng: Mutex<StdRng>,
 }
 
 impl LatencyIntercepor<Poisson<f32>> {
-    pub fn new_poisson(lambda_ms: f32) -> Result<Self, SimulationError> {
+    /// Creates a latency interceptor that samples delays from a Poisson distribution. If `seed` is provided the
+    /// sampled latencies are reproducible; otherwise the RNG is seeded from entropy.
+    pub fn new_poisson(lambda_ms: f32, seed: Option<u64>) -> Result<Self, SimulationError> {
         let poisson_dist = Poisson::new(lambda_ms).map_err(|e| {
             SimulationError::SimulatedNetworkError(format!("Could not create possion: {e}"))
         })?;
 
         Ok(Self {
             latency_dist: poisson_dist,
+            rng: Mutex::new(seeded_rng(seed)),
         })
     }
 }
 
+/// Builds an RNG from an optional seed, falling back to entropy when no seed is provided.
+fn seeded_rng(seed: Option<u64>) -> StdRng {
+    match seed {
+        Some(seed) => StdRng::seed_from_u64(seed),
+        None => StdRng::from_entropy(),
+    }
+}
+
 #[async_trait]
 impl<D> Interceptor for LatencyIntercepor<D>
 where
@@ -37,7 +53,13 @@ where
         &self,
         req: InterceptRequest,
     ) -> Result<Result<CustomRecords, ForwardingError>, CriticalError> {
-        let latency = self.latency_dist.sample(&mut rand::thread_rng());
+        let latency = {
+            let mut rng = self
+                .rng
+                .lock()
+                .expect("latency interceptor RNG lock poisoned");
+            self.latency_dist.sample(&mut *rng)
+        };
 
         select! {
             _ = req.shutdown_listener => log::debug!("Latency interceptor exiting due to shutdown signal received."),
@@ -62,7 +84,9 @@ mod tests {
     use lightning::ln::PaymentHash;
     use ntest::assert_true;
     use rand::distributions::Distribution;
-    use rand::Rng;
+    use rand::rngs::StdRng;
+    use rand::{Rng, SeedableRng};
+    use std::sync::Mutex;
     use tokio::time::timeout;
     use triggered::Trigger;
 
@@ -105,7 +129,10 @@ mod tests {
     async fn test_shutdown_signal() {
         // Set fixed dist to a high value so that the test won't flake.
         let latency_dist = ConstantDistribution { value: 1000.0 };
-        let interceptor = LatencyIntercepor { latency_dist };
+        let interceptor = LatencyIntercepor {
+            latency_dist,
+            rng: Mutex::new(StdRng::seed_from_u64(0)),
+        };
 
         let (request, trigger) = test_request();
         trigger.trigger();
@@ -121,7 +148,10 @@ mod tests {
     #[tokio::test]
     async fn test_latency_response() {
         let latency_dist = ConstantDistribution { value: 0.0 };
-        let interceptor = LatencyIntercepor { latency_dist };
+        let interceptor = LatencyIntercepor {
+            latency_dist,
+            rng: Mutex::new(StdRng::seed_from_u64(0)),
+        };
 
         let (request, _) = test_request();
         // We should return immediately because timeout is zero.

simln-lib/src/lib.rs

@@ -644,7 +644,12 @@ struct ExecutorPaymentTracker {
 
 impl Ord for PaymentEvent {
     fn cmp(&self, other: &Self) -> std::cmp::Ordering {
-        self.execution_time.cmp(&other.execution_time)
+        // Order primarily by execution time, then break ties on the source node's public key. The heap only ever holds
+        // at most one event per source at a time, so `(execution_time, source)` is a total order. Without the tie-break,
+        // events scheduled for the same instant pop in an unspecified order, which makes runs non-reproducible.
+        self.execution_time
+            .cmp(&other.execution_time)
+            .then_with(|| self.source.cmp(&other.source))
     }
 }
 
@@ -656,7 +661,7 @@ impl PartialOrd for PaymentEvent {
 
 impl PartialEq for PaymentEvent {
     fn eq(&self, other: &Self) -> bool {
-        self.execution_time == other.execution_time
+        self.execution_time == other.execution_time && self.source == other.source
     }
 }
 
@@ -1190,12 +1195,14 @@ async fn produce_payment_events<C: Clock>(
                             _ = pe_clock.sleep(wait_time) => {
                                 generate_payment(&mut heap, source, &mut payments_tracker, clock.now()).await?;
 
+                                // Stamp the dispatch time from the simulation clock now that the wait has elapsed.
+                                let dispatch_time = pe_clock.now();
                                 tasks.spawn(async move {
                                     log::debug!("Generated payment: {source} -> {}: {amount} msat.", destination);
 
                                     // Send the payment, exiting if we can no longer send to the consumer.
                                     let event = SimulationEvent::SendPayment(destination.clone(), amount);
-                                    if let Err(e) = send_payment(node, pe_output_sender, event.clone()).await {
+                                    if let Err(e) = send_payment(node, pe_output_sender, event.clone(), dispatch_time).await {
                                         pe_shutdown.trigger();
                                         log::debug!("Not able to send event payment for {amount}: {source} -> {}. Exited with error {e}.", destination);
                                     } else {
@@ -1270,6 +1277,7 @@ async fn send_payment(
     node: Arc<Mutex<dyn LightningNode>>,
     sender: Sender<SimulationOutput>,
     simulation_event: SimulationEvent,
+    dispatch_time: SystemTime,
 ) -> Result<(), SimulationError> {
     match simulation_event {
         SimulationEvent::SendPayment(dest, amt_msat) => {
@@ -1280,7 +1288,9 @@ async fn send_payment(
                 hash: None,
                 amount_msat: amt_msat,
                 destination: dest.pubkey,
-                dispatch_time: SystemTime::now(),
+                // Take the dispatch time from the simulation clock rather than the wall clock, so that it advances with
+                // virtual time under discrete-event simulation and stays reproducible across runs.
+                dispatch_time,
             };
 
             let outcome = match node.send_payment(dest.pubkey, amt_msat).await {
@@ -1509,6 +1519,7 @@ async fn produce_simulation_results(
                     },
                     SimulationOutput::SendPaymentFailure(payment, result) => {
                         select! {
+                            biased;
                             _ = listener.clone() => {
                                 return Ok(());
                             },
@@ -1669,6 +1680,39 @@ mod tests {
         assert_eq!(seq1, seq1_again);
     }
 
+    #[test]
+    fn test_payment_event_orders_ties_by_source() {
+        use crate::PaymentEvent;
+        use std::cmp::Reverse;
+        use std::collections::BinaryHeap;
+        use std::time::{Duration, SystemTime};
+
+        let nodes = test_utils::create_nodes(2, 100_000);
+        let (mut low, mut high) = (nodes[0].0.clone(), nodes[1].0.clone());
+        // Order our two nodes so that `low` has the smaller public key.
+        if low.pubkey > high.pubkey {
+            std::mem::swap(&mut low, &mut high);
+        }
+
+        let when = SystemTime::UNIX_EPOCH + Duration::from_secs(10);
+        let event = |source: &NodeInfo, destination: &NodeInfo| PaymentEvent {
+            source: source.pubkey,
+            execution_time: when,
+            destination: destination.clone(),
+            amount: 1_000,
+        };
+
+        // Push the higher-keyed source first to prove that pop order is decided by the tie-break, not by
+        // insertion order.
+        let mut heap: BinaryHeap<Reverse<PaymentEvent>> = BinaryHeap::new();
+        heap.push(Reverse(event(&high, &low)));
+        heap.push(Reverse(event(&low, &high)));
+
+        // Both events share an execution time, so the min-heap must pop the smaller public key first.
+        assert_eq!(heap.pop().unwrap().0.source, low.pubkey);
+        assert_eq!(heap.pop().unwrap().0.source, high.pubkey);
+    }
+
     mock! {
         pub Generator {}
 

simln-lib/src/sim_node.rs

@@ -964,6 +964,7 @@ async fn handle_intercepted_htlc(
     let mut interceptor_failure = None;
     'get_resp: loop {
         tokio::select! {
+            biased;
             res = intercepts.join_next() => {
                 let res = match res {
                     Some(res) => res,