fixup! 2beccfa

Author: f3r10

simln-lib/src/lib.rs

@@ -627,11 +627,15 @@ struct ExecutorKit {
 
 struct PaymentEvent {
     absolute_time: SystemTime,
+    executor: ExecutorKit,
+}
+
+#[derive(Clone)]
+struct PaymentEventPayload {
     wait_time: Duration,
     destination: NodeInfo,
-    current_count: u64,
     capacity: Option<u64>,
-    executor: ExecutorKit,
+    current_count: u64,
 }
 
 impl Ord for PaymentEvent {
@@ -1106,14 +1110,26 @@ impl<C: Clock + 'static> Simulation<C> {
         tasks: &TaskTracker,
     ) -> Result<(), SimulationError> {
         let mut heap: BinaryHeap<Reverse<PaymentEvent>> = BinaryHeap::new();
+        let payment_event_payloads: Arc<Mutex<HashMap<PublicKey, PaymentEventPayload>>> =
+            Arc::new(Mutex::new(HashMap::new()));
         for executor in executors {
-            generate_payment(&mut heap, executor, 0)?;
+            let payload = PaymentEventPayload {
+                wait_time: Duration::from_secs(0),
+                destination: executor.source_info.clone(),
+                capacity: None,
+                current_count: 0,
+            };
+            payment_event_payloads
+                .lock()
+                .await
+                .insert(executor.source_info.pubkey, payload);
+            generate_payment(&mut heap, executor, 0, payment_event_payloads.clone()).await?;
         }
 
         let listener = self.shutdown_listener.clone();
         let shutdown = self.shutdown_trigger.clone();
         let clock = self.clock.clone();
-        let t = tasks.clone();
+        let task_clone = tasks.clone();
 
         tasks.spawn(async move {
         loop {
@@ -1126,14 +1142,16 @@ impl<C: Clock + 'static> Simulation<C> {
                     match heap.pop() {
                         Some(Reverse(PaymentEvent {
                             absolute_time: _,
-                            wait_time,
-                            destination,
-                            current_count,
-                            capacity,
                             executor,
                         })) => {
+                            let payload = payment_event_payloads
+                                .lock()
+                                .await
+                                .get(&executor.source_info.pubkey)
+                                .ok_or(SimulationError::PaymentGenerationError(PaymentGenerationError(format!("executor {} not found", executor.source_info))))?
+                                .clone();
                             if let Some(c) = executor.payment_generator.payment_count() {
-                                if c == current_count {
+                                if c == payload.current_count {
                                     log::info!(
                                         "Payment count has been met for {}: {c} payments. Stopping the activity."
                                     , executor.source_info);
@@ -1157,14 +1175,14 @@ impl<C: Clock + 'static> Simulation<C> {
                             // Only proceed with a payment if the amount is non-zero, otherwise skip this round. If we can't get
                             // a payment amount something has gone wrong (because we should have validated that we can always
                             // generate amounts), so we exit.
-                            let payment_amount = executor.payment_generator.payment_amount(capacity);
+                            let payment_amount = executor.payment_generator.payment_amount(payload.capacity);
                             let amount = match payment_amount {
                                 Ok(amt) => {
                                     if amt == 0 {
                                         log::debug!(
-                                            "Skipping zero amount payment for {source} -> {destination}."
+                                            "Skipping zero amount payment for {source} -> {}.", payload.destination
                                         );
-                                        generate_payment(&mut heap, executor, current_count)?;
+                                        generate_payment(&mut heap, executor, payload.current_count, payment_event_payloads.clone()).await?;
                                         continue;
                                     }
                                     amt
@@ -1173,19 +1191,19 @@ impl<C: Clock + 'static> Simulation<C> {
                                     return Err(SimulationError::PaymentGenerationError(e));
                                 },
                             };
-                            generate_payment(&mut heap, executor, current_count + 1)?;
+                            generate_payment(&mut heap, executor, payload.current_count + 1, payment_event_payloads.clone()).await?;
 
                             // Wait until our time to next payment has elapsed then execute a random amount payment to a random
                             // destination.
-                            pe_clock.sleep(wait_time).await;
-                            t.spawn(async move {
-                                log::debug!("Generated payment: {source} -> {}: {amount} msat.", destination);
+                            pe_clock.sleep(payload.wait_time).await;
+                            task_clone.spawn(async move {
+                                log::debug!("Generated payment: {source} -> {}: {amount} msat.", payload.destination);
 
                                 // Send the payment, exiting if we can no longer send to the consumer.
-                                let event = SimulationEvent::SendPayment(destination.clone(), amount);
+                                let event = SimulationEvent::SendPayment(payload.destination.clone(), amount);
                                 if let Err(e) = pe_sender.send(event.clone()).await {
                                     pe_shutdown.trigger();
-                                    log::debug!("Not able to send event payment for {amount}: {source} -> {destination}. Exited with error {e}.");
+                                    log::debug!("Not able to send event payment for {amount}: {source} -> {}. Exited with error {e}.", payload.destination);
                                 } else {
 
                                     log::debug!("Send event payment for {source} completed successfully.");
@@ -1540,10 +1558,11 @@ async fn track_payment_result(
     Ok(())
 }
 
-fn generate_payment(
+async fn generate_payment(
     heap: &mut BinaryHeap<Reverse<PaymentEvent>>,
     executor: ExecutorKit,
     current_count: u64,
+    counters: Arc<Mutex<HashMap<PublicKey, PaymentEventPayload>>>,
 ) -> Result<(), SimulationError> {
     let now = SystemTime::now();
     let wait_time = get_payment_delay(
@@ -1560,12 +1579,24 @@ fn generate_payment(
         .network_generator
         .choose_destination(executor.source_info.pubkey)
         .map_err(SimulationError::DestinationGenerationError)?;
+    let mut payload = counters
+        .lock()
+        .await
+        .get(&executor.source_info.pubkey)
+        .ok_or(SimulationError::PaymentGenerationError(
+            PaymentGenerationError(format!("executor {} not found", executor.source_info)),
+        ))?
+        .clone();
+    payload.wait_time = wait_time;
+    payload.destination = destination;
+    payload.capacity = capacity;
+    payload.current_count = current_count;
+    counters
+        .lock()
+        .await
+        .insert(executor.source_info.pubkey, payload);
     let payment_event = PaymentEvent {
         absolute_time,
-        wait_time,
-        destination,
-        current_count,
-        capacity,
         executor,
     };
     heap.push(Reverse(payment_event));
@@ -2038,18 +2069,18 @@ mod tests {
             source: node_1.clone(),
             destination: node_2.clone(),
             start_secs: None,
-            count: Some(10),                               // No limit
-            interval_secs: crate::ValueOrRange::Value(5),  // 1 second interval
+            count: Some(10),                               // 10 payments
+            interval_secs: crate::ValueOrRange::Value(2),  // 2 second interval
             amount_msat: crate::ValueOrRange::Value(2000), // 2000 msats
         };
 
-        // Activity 2: From node_2 to node_1 - This would normally succeed
+        // Activity 2: From node_3 to node_4
         let activity_2 = crate::ActivityDefinition {
             source: node_3.clone(),
             destination: node_4.clone(),
-            start_secs: None, // Start 2 seconds after the first activity
-            count: Some(10),  // 10 payments if no error
-            interval_secs: crate::ValueOrRange::Value(10), // 1 second interval
+            start_secs: None,
+            count: Some(10),                               // 10 payments
+            interval_secs: crate::ValueOrRange::Value(4),  // 4 second interval
             amount_msat: crate::ValueOrRange::Value(3000), // 3000 msats
         };
 
@@ -2058,7 +2089,7 @@ mod tests {
         // Create simulation with a long timeout that we don't expect to be reached
         let simulation = Simulation::new(
             SimulationCfg::new(
-                None,     // 30 second timeout (shouldn't matter)
+                None,     // without timeout
                 100,      // Expected payment size
                 2.0,      // Activity multiplier
                 None,     // No result writing
@@ -2076,12 +2107,12 @@ mod tests {
         let _ = simulation.run(&vec![activity_1, activity_2]).await;
         let elapsed = start.elapsed();
 
-        // Check that simulation ran 150 because
-        // from activity_1 there are 10 payments with a wait_time of 5s
-        // from activity_2 there are 10 payments with a wait_time of 10s
+        // Check that simulation ran 60 because
+        // from activity_1 there are 10 payments with a wait_time of 2s -> 20s
+        // from activity_2 there are 10 payments with a wait_time of 4s -> 40s
         assert!(
-            elapsed >= Duration::from_secs(150),
-            "Simulation should have run at least for 150s, took {:?}",
+            elapsed >= Duration::from_secs(60),
+            "Simulation should have run at least for 54s, took {:?}",
             elapsed
         );