feat: add optional name for activity descriptions

Author: Ifeanyichukwu

sim-cli/src/main.rs

@@ -95,6 +95,7 @@ async fn main() -> anyhow::Result<()> {
     let mut clients: HashMap<PublicKey, Arc<Mutex<dyn LightningNode>>> = HashMap::new();
     let mut pk_node_map = HashMap::new();
     let mut alias_node_map = HashMap::new();
+    let mut activity_name_map = HashMap::new();
 
     for connection in nodes {
         // TODO: Feels like there should be a better way of doing this without having to Arc<Mutex<T>>> it at this time.
@@ -182,13 +183,23 @@ async fn main() -> anyhow::Result<()> {
             },
         };
 
+        if activity_name_map.contains_key(&act.activity_name) {
+            anyhow::bail!(LightningError::ValidationError(format!(
+                "Duplicate activity name {:?} is not allowed.",
+                act.activity_name.unwrap()
+            )));
+        }
+
+        activity_name_map.insert(act.activity_name.clone(), act.clone());
+
         validated_activities.push(ActivityDefinition {
             source,
             destination,
             start_secs: act.start_secs,
             count: act.count,
             interval_secs: act.interval_secs,
             amount_msat: act.amount_msat,
+            activity_name: act.activity_name,
         });
     }
 

sim-lib/src/lib.rs

@@ -149,6 +149,8 @@ pub struct ActivityParser {
     pub interval_secs: u16,
     /// The amount of m_sat to used in this payment.
     pub amount_msat: u64,
+    /// an optional name for the activity.
+    pub activity_name: Option<String>,
 }
 
 /// Data structure used internally by the simulator. Both source and destination are represented as [PublicKey] here.
@@ -167,6 +169,8 @@ pub struct ActivityDefinition {
     pub interval_secs: u16,
     /// The amount of m_sat to used in this payment.
     pub amount_msat: u64,
+    /// An optional name for the activity.
+    pub activity_name: Option<String>,
 }
 
 #[derive(Debug, Error)]
@@ -336,14 +340,16 @@ pub enum PaymentOutcome {
 }
 
 /// Describes a payment from a source node to a destination node.
-#[derive(Debug, Clone, Copy, Serialize)]
+#[derive(Debug, Clone, Serialize)]
 struct Payment {
     /// Pubkey of the source node dispatching the payment.
     source: PublicKey,
     /// Pubkey of the destination node receiving the payment.
     destination: PublicKey,
     /// Amount of the payment in msat.
     amount_msat: u64,
+    /// Name of the activity.
+    activity_name: String,
     /// Hash of the payment if it has been successfully dispatched.
     #[serde(with = "serializers::serde_option_payment_hash")]
     hash: Option<PaymentHash>,
@@ -352,6 +358,16 @@ struct Payment {
     dispatch_time: SystemTime,
 }
 
+impl Payment {
+    fn formatted_activity_name(&self) -> String {
+        if !self.activity_name.is_empty() {
+            format!("{} activity:", self.activity_name)
+        } else {
+            "".to_string()
+        }
+    }
+}
+
 impl Display for Payment {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         let dispatch_time = self
@@ -361,7 +377,8 @@ impl Display for Payment {
 
         write!(
             f,
-            "Payment {} dispatched at {:?} sending {} msat from {} -> {}.",
+            "{} Payment {} dispatched at {:?} sending {} msat from {} -> {}.",
+            self.formatted_activity_name(),
             self.hash.map(|h| hex::encode(h.0)).unwrap_or_default(),
             dispatch_time,
             self.amount_msat,
@@ -377,7 +394,7 @@ impl Display for Payment {
 enum SimulationEvent {
     /// Dispatch a payment of the specified amount to the public key provided.
     /// Results in `SimulationOutput::SendPaymentSuccess` or `SimulationOutput::SendPaymentFailure`.
-    SendPayment(NodeInfo, u64),
+    SendPayment(NodeInfo, u64, String),
 }
 
 /// SimulationOutput provides the output of a simulation event.
@@ -427,6 +444,7 @@ struct ExecutorKit {
     /// See [NetworkGraphView] for details.
     network_generator: Arc<Mutex<dyn DestinationGenerator>>,
     payment_generator: Box<dyn PaymentGenerator>,
+    activity_name: String,
 }
 
 impl Simulation {
@@ -696,7 +714,12 @@ impl Simulation {
         // Note: when we allow configuring both defined and random activity, this will no longer be an if/else, we'll
         // just populate with each type as configured.
         if !self.activity.is_empty() {
-            for description in self.activity.iter() {
+            for (index, description) in self.activity.iter().enumerate() {
+                let activity_name = match &description.activity_name {
+                    Some(name) => name.clone(),
+                    None => format!("Index {}", index),
+                };
+
                 let activity_generator = DefinedPaymentActivity::new(
                     description.destination.clone(),
                     Duration::from_secs(description.start_secs.into()),
@@ -711,6 +734,7 @@ impl Simulation {
                     // a single struct which we just cheaply clone.
                     network_generator: Arc::new(Mutex::new(activity_generator.clone())),
                     payment_generator: Box::new(activity_generator),
+                    activity_name,
                 });
             }
         } else {
@@ -762,6 +786,7 @@ impl Simulation {
             generators.push(ExecutorKit {
                 source_info: node_info.clone(),
                 network_generator: network_generator.clone(),
+                activity_name: "".to_string(),
                 payment_generator: Box::new(
                     RandomPaymentActivity::new(
                         *capacity,
@@ -845,13 +870,15 @@ impl Simulation {
                 let source = executor.source_info.clone();
 
                 log::info!(
-                    "Starting activity producer for {}: {}.",
+                    "Starting {} activity producer for {}: {}.",
+                    executor.activity_name,
                     source,
                     executor.payment_generator
                 );
 
                 if let Err(e) = produce_events(
                     executor.source_info,
+                    executor.activity_name,
                     executor.network_generator,
                     executor.payment_generator,
                     pe_sender,
@@ -888,7 +915,7 @@ async fn consume_events(
             simulation_event = receiver.recv() => {
                 if let Some(event) = simulation_event {
                     match event {
-                        SimulationEvent::SendPayment(dest, amt_msat) => {
+                        SimulationEvent::SendPayment(dest, amt_msat, activity_name) => {
                             let mut node = node.lock().await;
 
                             let mut payment = Payment {
@@ -897,6 +924,7 @@ async fn consume_events(
                                 amount_msat: amt_msat,
                                 destination: dest.pubkey,
                                 dispatch_time: SystemTime::now(),
+                                activity_name,
                             };
 
                             let outcome = match node.send_payment(dest.pubkey, amt_msat).await {
@@ -947,6 +975,7 @@ async fn consume_events(
 /// exit if other threads signal that they have errored out.
 async fn produce_events<N: DestinationGenerator + ?Sized, A: PaymentGenerator + ?Sized>(
     source: NodeInfo,
+    activity_name: String,
     network_generator: Arc<Mutex<N>>,
     node_generator: Box<A>,
     sender: Sender<SimulationEvent>,
@@ -996,7 +1025,7 @@ async fn produce_events<N: DestinationGenerator + ?Sized, A: PaymentGenerator +
                 let amount = match node_generator.payment_amount(capacity) {
                     Ok(amt) => {
                         if amt == 0 {
-                            log::debug!("Skipping zero amount payment for {source} -> {destination}.");
+                            log::debug!("{activity_name} activity : Skipping zero amount payment for {source} -> {destination}.");
                             continue;
                         }
                         amt
@@ -1009,7 +1038,7 @@ async fn produce_events<N: DestinationGenerator + ?Sized, A: PaymentGenerator +
                 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);
+                let event = SimulationEvent::SendPayment(destination.clone(), amount, activity_name.clone());
                 if sender.send(event.clone()).await.is_err() {
                     return Err(SimulationError::MpscChannelError (format!("Stopped activity producer for {amount}: {source} -> {destination}.")));
                 }
@@ -1076,11 +1105,12 @@ async fn consume_simulation_results(
 }
 
 /// PaymentResultLogger is an aggregate logger that will report on a summary of the payments that have been reported.
-#[derive(Default)]
+#[derive(Default, Clone)]
 struct PaymentResultLogger {
     success_payment: u64,
     failed_payment: u64,
     total_sent: u64,
+    activity_name: String,
 }
 
 impl PaymentResultLogger {
@@ -1097,6 +1127,7 @@ impl PaymentResultLogger {
         }
 
         self.total_sent += details.amount_msat;
+        self.activity_name = details.formatted_activity_name();
     }
 }
 
@@ -1105,7 +1136,8 @@ impl Display for PaymentResultLogger {
         let total_payments = self.success_payment + self.failed_payment;
         write!(
             f,
-            "Processed {} payments sending {} msat total with {:.2}% success rate.",
+            "{} Processed {} payments sending {} msat total with {:.2}% success rate.",
+            self.activity_name,
             total_payments,
             self.total_sent,
             (self.success_payment as f64 / total_payments as f64) * 100.0
@@ -1174,7 +1206,7 @@ async fn produce_simulation_results(
                                 }
                             },
                             SimulationOutput::SendPaymentFailure(payment, result) => {
-                                if results.send((payment, result.clone())).await.is_err() {
+                                if results.send((payment.clone(), result.clone())).await.is_err() {
                                     return Err(SimulationError::MpscChannelError(
                                         format!("Failed to send payment result: {result} for payment {:?} dispatched at {:?}.", payment.hash, payment.dispatch_time),
                                     ));
@@ -1245,7 +1277,7 @@ async fn track_payment_result(
         _ = listener.clone() => {
             log::debug!("Track payment result received a shutdown signal.");
         },
-        send_payment_result = results.send((payment, res.clone())) => {
+        send_payment_result = results.send((payment.clone(), res.clone())) => {
             if send_payment_result.is_err() {
                 return Err(SimulationError::MpscChannelError(format!("Failed to send payment result {res} for payment {payment}.")))
             }