diff --git a/sim-cli/src/main.rs b/sim-cli/src/main.rs index e6c7cf9d..6f698746 100755 --- a/sim-cli/src/main.rs +++ b/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] + vec![Arc::new(LatencyIntercepor::new_poisson( + latency as f32, + cli.fix_seed, + )?) as Arc] } else { vec![] }; diff --git a/simln-lib/src/latency_interceptor.rs b/simln-lib/src/latency_interceptor.rs index 9034b8b1..3d904972 100644 --- a/simln-lib/src/latency_interceptor.rs +++ b/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 + 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, } impl LatencyIntercepor> { - pub fn new_poisson(lambda_ms: f32) -> Result { + /// 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) -> Result { 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) -> StdRng { + match seed { + Some(seed) => StdRng::seed_from_u64(seed), + None => StdRng::from_entropy(), + } +} + #[async_trait] impl Interceptor for LatencyIntercepor where @@ -37,7 +53,13 @@ where &self, req: InterceptRequest, ) -> Result, 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. diff --git a/simln-lib/src/lib.rs b/simln-lib/src/lib.rs index a1b5fe0e..ca30f4be 100755 --- a/simln-lib/src/lib.rs +++ b/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( _ = 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>, sender: Sender, 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> = 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 {} diff --git a/simln-lib/src/sim_node.rs b/simln-lib/src/sim_node.rs index 801315f1..d0940798 100755 --- a/simln-lib/src/sim_node.rs +++ b/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,