diff --git a/README.md b/README.md index ea3358f2..31887a51 100644 --- a/README.md +++ b/README.md @@ -377,6 +377,64 @@ This is required for the case where you want to specify a channel that will be used in the simulated network that is just responsible for forwarding payments, but does not send or receive any payments itself. +### Channel Rebalancing + +Since payment flows in the simulation are not perfectly circular, the +channels in a simulated network will progressively deplete to one side +over the course of a long-running simulation, and payment success rates +will decay accordingly. On the real network, node operators counteract +this with out-of-band actions such as circular rebalances, swaps and +splices. The simulator can optionally model the *effect* of these +actions (without their mechanics) by periodically "teleporting" the +balance of depleted channels back to an even split: + +``` +{ + "sim_network": [ ... ], + "rebalance": { + "trigger_below": 0.1, + "interval_secs": 3600, + "exclude": [ + "020a30431ce58843eedf8051214dbfadb65b107cc598b8277f14bb9b33c9cd026f" + ] + } +} +``` + +* `trigger_below`: the fraction of a channel's capacity beneath which + either side's available balance triggers a rebalance of the channel, + exclusively between 0 and 0.5 (default: 0.1). +* `interval_secs`: the number of seconds between scans of the network's + channels (default: 3600). +* `exclude`: nodes that do not participate in rebalancing (default: + empty). Any channel that has a listed node as one of its peers will + never be rebalanced, regardless of which side of the channel has + depleted. This models operators that don't maintain their liquidity, + letting their channels drain naturally while the rest of the network + is kept liquid. + +Rebalancing is opt-in and only available on simulated networks. Leave +it off to study how liquidity depletes under a traffic pattern; turn it +on for long-running simulations that need sustained payment success +(each rebalance is logged with the amount moved, so the volume of +intervention required to keep the network liquid is itself a useful +output). Rebalancing preserves determinism when running with a fixed +seed. Balance that is locked in in-flight HTLCs is never moved. + +### Routing Fee Limit + +Payments on simulated networks will not pay more than 5% of the +payment amount in routing fees, mirroring the fee limits that real +senders place on payments. This can be tuned with a top-level field in +the simulation file (only valid on simulated networks): + +``` +{ + "sim_network": [ ... ], + "max_route_fee_pct": 0.5 +} +``` + ### Inclusions and Limitations The simulator will execute payments on the mocked out network as it diff --git a/sim-cli/src/parsing.rs b/sim-cli/src/parsing.rs index c5b06b7f..5e9605a8 100755 --- a/sim-cli/src/parsing.rs +++ b/sim-cli/src/parsing.rs @@ -6,7 +6,7 @@ use serde::{Deserialize, Serialize}; use simln_lib::clock::SimulationClock; use simln_lib::sim_node::{ ln_node_from_graph, populate_network_graph, ChannelPolicy, CustomRecords, Interceptor, - SimGraph, SimNode, SimulatedChannel, + RebalanceConfig, SimGraph, SimNode, SimulatedChannel, DEFAULT_MAX_ROUTE_FEE_PCT, }; use simln_lib::{ cln, cln::ClnNode, eclair, eclair::EclairNode, lnd, lnd::LndNode, serializers, @@ -14,10 +14,11 @@ use simln_lib::{ Simulation, SimulationCfg, WriteResults, }; use simln_lib::{ShortChannelID, SimulationError}; -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::fs; use std::path::PathBuf; use std::sync::Arc; +use std::time::Duration; use tokio::sync::Mutex; use tokio_util::task::TaskTracker; @@ -138,6 +139,18 @@ impl Cli { } } + if sim_params.rebalance.is_some() && sim_params.sim_network.is_empty() { + return Err(anyhow!( + "Rebalancing is only allowed on a simulated network" + )); + } + + if sim_params.max_route_fee_pct.is_some() && sim_params.sim_network.is_empty() { + return Err(anyhow!( + "A routing fee limit can only be set on a simulated network" + )); + } + Ok(()) } } @@ -152,6 +165,42 @@ pub struct SimParams { pub activity: Vec, #[serde(default)] pub exclude: Vec, + #[serde(default)] + pub rebalance: Option, + /// The maximum total routing fee that simulated nodes will pay for a payment, expressed as a percentage of + /// the payment amount (only used on simulated networks, where it defaults to 5%). + #[serde(default)] + pub max_route_fee_pct: Option, +} + +/// Default fraction of channel capacity beneath which a channel side triggers a rebalance. +fn default_rebalance_trigger() -> f64 { + 0.1 +} + +/// Default number of seconds between rebalancing scans of the simulated network (one hour). +fn default_rebalance_interval() -> u64 { + 3600 +} + +/// Data structure that is used to parse rebalancing configuration for simulated networks from the simulation +/// file. When present, channels in the simulated network whose available liquidity has drawn down beneath a +/// trigger fraction of their capacity are periodically reset to an even split, modeling the out-of-band actions +/// (circular rebalances, swaps, splices) that node operators take to restore liquidity. When absent, channels +/// deplete naturally under the simulation's payment flows. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct RebalanceParser { + /// The fraction of a channel's capacity beneath which a side's available balance triggers a rebalance, + /// exclusively between 0 and 0.5. + #[serde(default = "default_rebalance_trigger")] + pub trigger_below: f64, + /// The number of seconds between scans of the network's channels. + #[serde(default = "default_rebalance_interval")] + pub interval_secs: u64, + /// Nodes that do not participate in rebalancing. Any channel that has a node in this list as one of its + /// peers will never be rebalanced, regardless of which side of the channel has depleted. + #[serde(default)] + pub exclude: Vec, } #[derive(Serialize, Deserialize, Debug, Clone)] @@ -260,6 +309,8 @@ pub async fn create_simulation_with_network( sim_network, activity: _activity, exclude, + rebalance, + max_route_fee_pct, } = sim_params; // Convert nodes representation for parsing to SimulatedChannel @@ -300,6 +351,32 @@ pub async fn create_simulation_with_network( .map_err(|e| SimulationError::SimulatedNetworkError(format!("{:?}", e)))?, )); + // If the user has opted into rebalancing, spawn a task that will periodically reset channels whose + // liquidity has drawn down. Without this, the network's channels will progressively deplete under the + // simulation's payment flows. + if let Some(rebalance) = rebalance { + // Fail early on exclusions for nodes that aren't part of the simulated network, because they'd + // otherwise be silently ignored (most likely hiding a typo in the pubkey). + for pk in &rebalance.exclude { + if !nodes_info.contains_key(pk) { + return Err(anyhow!( + "Rebalance exclusion pubkey: {pk} not found in simulated network" + )); + } + } + + let rebalance_cfg = RebalanceConfig::new( + rebalance.trigger_below, + Duration::from_secs(rebalance.interval_secs), + HashSet::from_iter(rebalance.exclude.iter().copied()), + )?; + + simulation_graph + .lock() + .await + .start_rebalancer(clock.clone(), rebalance_cfg); + } + // Copy all simulated channels into a read-only routing graph, allowing to pathfind for // individual payments without locking the simulation graph (this is a duplication of the channels, // but the performance tradeoff is worthwhile for concurrent pathfinding). @@ -312,7 +389,13 @@ pub async fn create_simulation_with_network( // custom actions on the simulated network. For the nodes we'll pass our simulation, cast them // to a dyn trait and exclude any nodes that shouldn't be included in random activity // generation. - let nodes = ln_node_from_graph(simulation_graph, routing_graph, clock.clone()).await?; + let nodes = ln_node_from_graph( + simulation_graph, + routing_graph, + clock.clone(), + max_route_fee_pct.unwrap_or(DEFAULT_MAX_ROUTE_FEE_PCT), + ) + .await?; let mut nodes_dyn: HashMap<_, Arc>> = nodes .iter() .map(|(pk, node)| (*pk, Arc::clone(node) as Arc>)) @@ -351,6 +434,8 @@ pub async fn create_simulation( sim_network: _sim_network, activity: _activity, exclude: _, + rebalance: _, + max_route_fee_pct: _, } = sim_params; let (clients, clients_info) = get_clients(nodes.to_vec()).await?; diff --git a/simln-lib/src/lib.rs b/simln-lib/src/lib.rs index ca30f4be..3a7ed97c 100755 --- a/simln-lib/src/lib.rs +++ b/simln-lib/src/lib.rs @@ -102,7 +102,7 @@ impl std::fmt::Display for NodeId { } /// Represents a short channel ID, expressed as a struct so that we can implement display for the trait. -#[derive(Debug, Clone, PartialEq, Eq, Hash, Copy, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Copy, Serialize, Deserialize)] pub struct ShortChannelID(u64); /// Utility function to easily convert from u64 to `ShortChannelID`. @@ -2162,15 +2162,23 @@ mod tests { "Simulation should have run at least for 25s, took {:?}", elapsed ); + // The total number of payments made within the simulation's 25 second runtime depends on wall clock + // scheduling at the cutoff, so we only assert on a prefix of payments that will reliably complete in time. let expected_payment_list = vec![ - pk1, pk2, pk1, pk1, pk1, pk3, pk3, pk3, pk4, pk3, pk2, pk1, pk4, + pk1, pk2, pk1, pk1, pk3, pk3, pk3, pk4, pk2, pk4, pk3, pk2, pk3, pk2, pk4, pk4, pk1, + pk4, pk2, pk4, ]; - assert!( - payments_list.lock().unwrap().as_ref() == expected_payment_list, - "The expected order of payments is not correct: {:?} vs {:?}", - payments_list.lock().unwrap(), - expected_payment_list, + let actual_payments: Vec = payments_list + .lock() + .unwrap() + .iter() + .take(expected_payment_list.len()) + .cloned() + .collect(); + assert_eq!( + actual_payments, expected_payment_list, + "The expected order of payments is not correct" ); // remove all the payments made in the previous execution @@ -2189,9 +2197,16 @@ mod tests { ); let _ = simulation2.run(&[]).await; - assert!( - payments_list.lock().unwrap().as_ref() != expected_payment_list, - "The expected order of payments shoud be different because a different is used" + let actual_payments: Vec = payments_list + .lock() + .unwrap() + .iter() + .take(expected_payment_list.len()) + .cloned() + .collect(); + assert_ne!( + actual_payments, expected_payment_list, + "The expected order of payments should be different because a different seed is used" ); } } diff --git a/simln-lib/src/random_activity.rs b/simln-lib/src/random_activity.rs index 466981b1..13bc76ff 100644 --- a/simln-lib/src/random_activity.rs +++ b/simln-lib/src/random_activity.rs @@ -14,6 +14,17 @@ use crate::{ const HOURS_PER_MONTH: u64 = 30 * 24; const SECONDS_PER_MONTH: u64 = HOURS_PER_MONTH * 60 * 60; +/// The 95th percentile (z-score) of the standard normal distribution. Payment amount distributions are chosen such +/// that at least 95% of sampled amounts fall below the payment limit for the node pair. +const PAYMENT_LIMIT_QUANTILE: f64 = 1.6449; + +/// The maximum sigma for the log normal distribution that payment amounts are sampled from, sqrt(ln(2)), which caps +/// the distribution's coefficient of variation (standard deviation / mean) at one. Bounding sigma keeps the tail of +/// the distribution light enough that the average of sampled payment amounts converges on the expected payment +/// amount within a reasonable number of payments (the standard error of the mean over n payments is at most +/// expected_payment_amt / sqrt(n)). +const MAX_SIGMA: f64 = 0.8325546111576977; + #[derive(Debug, Error)] pub enum RandomActivityError { #[error("Value error: {0}")] @@ -183,16 +194,12 @@ impl RandomPaymentActivity { node_capacity_msat: u64, expected_payment_amt: u64, ) -> Result<(), RandomActivityError> { - // We will not be able to generate payments if the variance of sigma squared for our log normal distribution - // is < 0 (because we have to take a square root). - // - // Sigma squared is calculated as: 2(ln(payment_limit) - ln(expected_payment_amt)) - // Where: payment_limit = node_capacity_msat / 2. + // We will not be able to generate payments if the payment limit for a node pair (half of the smaller node's + // capacity) is beneath the expected payment amount, because no log normal distribution has the expected + // amount as its mean while keeping 95% of its samples below a limit that is smaller than that mean. // - // Therefore we can only process payments if: 2(ln(payment_limit) - ln(expected_payment_amt)) >= 0 - // ln(payment_limit) >= ln(expected_payment_amt) - // e^(ln(payment_limit) >= e^(ln(expected_payment_amt)) - // payment_limit >= expected_payment_amt + // The payment limit is at most node_capacity_msat / 2 (it may be smaller, depending on the capacity of the + // destination chosen per-payment), so we require: // node_capacity_msat / 2 >= expected_payment_amt // node_capacity_msat >= 2 * expected_payment_amt let min_required_capacity = 2 * expected_payment_amt; @@ -245,13 +252,13 @@ impl PaymentGenerator for RandomPaymentActivity { Ok(Duration::from_secs(duration_in_secs)) } - /// Returns the payment amount for a payment to a node with the destination capacity provided. The expected value - /// for the payment is the simulation expected payment amount, and the variance is determined by the channel - /// capacity of the source and destination node. Variance is calculated such that 95% of payment amounts generated - /// will fall between the expected payment amount and 50% of the capacity of the node with the least channel - /// capacity. While the expected value of payments remains the same, scaling variance by node capacity means that - /// nodes with more deployed capital will see a larger range of payment values than those with smaller total - /// channel capacity. + /// Returns the payment amount for a payment to a node with the destination capacity provided. Amounts are + /// sampled from a log normal distribution with mean equal to the simulation's expected payment amount, and + /// variance chosen per node-pair such that at least 95% of sampled amounts fall below a payment limit of 50% + /// of the capacity of the node with the least channel capacity. Node pairs with more deployed capital therefore + /// see a wider range of payment amounts, while the average amount remains the expected payment amount, so the + /// volume that a node sends over time converges on the total implied by its capacity and the simulation's + /// activity multiplier. fn payment_amount( &self, destination_capacity: Option, @@ -262,20 +269,35 @@ impl PaymentGenerator for RandomPaymentActivity { let payment_limit = std::cmp::min(self.source_capacity, destination_capacity) / 2; - let ln_pmt_amt = (self.expected_payment_amt as f64).ln(); - let ln_limit = (payment_limit as f64).ln(); - - let mu = 2.0 * ln_pmt_amt - ln_limit; - let sigma_square = 2.0 * (ln_limit - ln_pmt_amt); - - if sigma_square < 0.0 { + // A log normal distribution with mu = ln(mean) - sigma^2 / 2 has the expected payment amount as its mean + // for any choice of sigma. We pick sigma as large as possible for the node pair, subject to two constraints: + // * At least 95% of sampled amounts fall below the payment limit, which requires: + // (ln(limit) - mu) / sigma >= z, ie: sigma^2 - 2 * z * sigma + 2 * ln(limit / mean) >= 0. + // * Sigma may not exceed MAX_SIGMA, so that the tail of the distribution stays light enough for sample + // averages to converge on the mean quickly. + // + // When the quadratic's discriminant is negative, the limit constraint holds for any sigma (the limit is far + // above the mean); otherwise sigma must fall below the quadratic's smaller root. The region above the larger + // root also satisfies the constraint, but produces a heavy-tailed distribution where most payments are tiny + // and the volume target is carried by extremely rare, large payments. + let ln_ratio = (payment_limit as f64 / self.expected_payment_amt as f64).ln(); + if ln_ratio < 0.0 { return Err(PaymentGenerationError(format!( - "payment amount not possible for limit: {payment_limit}, sigma squared: {sigma_square}" + "expected payment amount: {} exceeds payment limit: {payment_limit}", + self.expected_payment_amt ))); } - let log_normal = LogNormal::new(mu, sigma_square.sqrt()) - .map_err(|e| PaymentGenerationError(e.to_string()))?; + let discriminant = PAYMENT_LIMIT_QUANTILE * PAYMENT_LIMIT_QUANTILE - 2.0 * ln_ratio; + let sigma = if discriminant <= 0.0 { + MAX_SIGMA + } else { + MAX_SIGMA.min(PAYMENT_LIMIT_QUANTILE - discriminant.sqrt()) + }; + let mu = (self.expected_payment_amt as f64).ln() - sigma * sigma / 2.0; + + let log_normal = + LogNormal::new(mu, sigma).map_err(|e| PaymentGenerationError(e.to_string()))?; let mut rng = self .rng @@ -457,10 +479,12 @@ mod tests { #[test] fn test_payment_amount() { - // The special cases for payment_amount are those who may make the internal log normal distribution fail to build, which happens if - // sigma squared is either +-INF or NaN. Given that the constructor of the PaymentActivityGenerator already forces its internal values - // to be greater than zero, the only values that are left are all values of `destination_capacity` smaller or equal to the `source_capacity` - // All of them will yield a sigma squared smaller than 0, which we have a sanity check for. + // Payment amounts can only be generated if the payment limit for the node pair (half of the smaller + // node's capacity) is at least the expected payment amount, because no log normal distribution has the + // expected amount as its mean while keeping 95% of samples below a limit that is smaller than that mean. + // Given that the constructor of the PaymentActivityGenerator already forces its internal values to be + // greater than zero, the only values that are left are all values of `destination_capacity` small enough + // to drag the payment limit beneath the expected payment amount, which we have a sanity check for. let expected_payment = get_random_int(1, 100); let source_capacity = 2 * expected_payment; let rng = MutRng::new(Some((u64::MAX, None))); @@ -490,5 +514,73 @@ mod tests { Err(PaymentGenerationError(..)) )); } + + #[test] + fn test_payment_amount_mean_convergence() { + // The average of generated payment amounts should converge on the expected payment amount within a + // reasonable number of payments, otherwise nodes do not send the volume that their capacity and the + // simulation's activity multiplier imply. Capacities are much larger than the expected payment amount + // so that sigma is at its cap; with a coefficient of variation of at most one, the standard error of + // the mean over 1000 samples is ~3% of the expected amount, so a 15% bound has comfortable headroom + // while still catching heavy-tailed parametrizations (which produce sample means that are off by + // orders of magnitude). + let expected_payment = 3_800_000; + let capacity = 1_000_000_000_000; + let rng = MutRng::new(Some((42, None))); + let pag = RandomPaymentActivity::new(capacity, expected_payment, 1.0, rng).unwrap(); + + let n = 1000; + let sum: u64 = (0..n) + .map(|_| pag.payment_amount(Some(capacity)).unwrap()) + .sum(); + let mean = sum as f64 / n as f64; + + assert!( + (mean - expected_payment as f64).abs() < 0.15 * expected_payment as f64, + "sample mean {mean} not within 15% of expected payment amount {expected_payment}", + ); + } + + #[test] + fn test_payment_amount_respects_limit() { + // At least 95% of payment amounts should fall below the payment limit of half of the smaller node's + // capacity. The destination is small relative to the source so that the limit constraint (rather than + // the cap on sigma) binds. + let expected_payment = 3_800_000; + let rng = MutRng::new(Some((42, None))); + let pag = + RandomPaymentActivity::new(1_000_000_000_000, expected_payment, 1.0, rng).unwrap(); + + let destination_capacity = 10 * expected_payment; + let limit = destination_capacity / 2; + let n = 1000; + let below = (0..n) + .filter(|_| pag.payment_amount(Some(destination_capacity)).unwrap() <= limit) + .count(); + + assert!( + below >= 950, + "expected at least 950/1000 payment amounts below limit {limit}, got {below}", + ); + } + + #[test] + fn test_payment_amount_degenerate_limit() { + // When the payment limit is exactly the expected payment amount, the only distribution that has the + // expected amount as its mean and respects the limit is a constant, so every payment should be the + // expected amount (within one msat of floating point truncation). + let expected_payment = 3_800_000; + let rng = MutRng::new(Some((42, None))); + let pag = RandomPaymentActivity::new(2 * expected_payment, expected_payment, 1.0, rng) + .unwrap(); + + for _ in 0..10 { + let amt = pag.payment_amount(Some(2 * expected_payment)).unwrap(); + assert!( + amt.abs_diff(expected_payment) <= 1, + "expected constant payment of {expected_payment}, got {amt}", + ); + } + } } } diff --git a/simln-lib/src/sim_node.rs b/simln-lib/src/sim_node.rs index d0940798..67bde288 100755 --- a/simln-lib/src/sim_node.rs +++ b/simln-lib/src/sim_node.rs @@ -8,7 +8,7 @@ use bitcoin::secp256k1::PublicKey; use bitcoin::{Network, ScriptBuf, TxOut}; use lightning::ln::chan_utils::make_funding_redeemscript; use serde::{Deserialize, Serialize}; -use std::collections::{hash_map::Entry, HashMap}; +use std::collections::{hash_map::Entry, HashMap, HashSet}; use std::fmt::Display; use std::sync::Arc; use std::time::UNIX_EPOCH; @@ -27,6 +27,7 @@ use lightning::routing::scoring::{ }; use lightning::routing::utxo::{UtxoLookup, UtxoResult}; use lightning::util::logger::{Level, Logger, Record}; +use std::time::Duration; use thiserror::Error; use tokio::select; use tokio::sync::oneshot::{channel, Receiver, Sender}; @@ -414,6 +415,33 @@ impl SimulatedChannel { Ok(()) } + /// "Teleports" the channel's settled balance back to an even split between its two nodes if either side's + /// available balance has dropped beneath `trigger_below_fraction` of the channel's total capacity, returning + /// the amount moved if the channel was rebalanced. This models the out-of-band actions that node operators + /// take to restore usable liquidity (circular rebalances, swaps, splices) by their effect rather than their + /// mechanism. + /// + /// Only the settled portion of the channel's capacity is redistributed; balance locked in in-flight HTLCs is + /// left untouched and will settle to one side or the other as usual, preserving the channel's accounting + /// invariant (the sides' local balances and in-flight totals always sum to the channel capacity). + fn rebalance_if_depleted(&mut self, trigger_below_fraction: f64) -> Option { + let threshold = (self.capacity_msat as f64 * trigger_below_fraction) as u64; + if self.node_1.local_balance_msat >= threshold + && self.node_2.local_balance_msat >= threshold + { + return None; + } + + let settled = self.node_1.local_balance_msat + self.node_2.local_balance_msat; + let target_1 = settled / 2; + let moved = self.node_1.local_balance_msat.abs_diff(target_1); + + self.node_1.local_balance_msat = target_1; + self.node_2.local_balance_msat = settled - target_1; + + Some(moved) + } + /// Removes an htlc from the appropriate side of the simulated channel, settling balances across channel sides /// based on the success of the htlc. The public key of the node that originally sent the HTLC (ie, the party /// that would send update_add_htlc in the protocol) must be provided to remove the htlc from its side of the @@ -524,17 +552,28 @@ pub struct SimNode { scorer: Mutex, Arc>>, /// Clock for tracking simulation time. clock: Arc, + /// The maximum total routing fee the node will pay for a payment, as a percentage of the payment amount. + max_route_fee_pct: f64, } impl SimNode { /// Creates a new simulation node that refers to the high level network coordinator provided to process payments /// on its behalf. The pathfinding graph is provided separately so that each node can handle its own pathfinding. + /// The node will not pay more than `max_route_fee_pct` percent of a payment's amount in routing fees, which + /// must be greater than zero. pub fn new( info: NodeInfo, payment_network: Arc>, pathfinding_graph: Arc, clock: Arc, + max_route_fee_pct: f64, ) -> Result { + if max_route_fee_pct <= 0.0 { + return Err(LightningError::ValidationError(format!( + "max_route_fee_pct must be greater than zero, got: {max_route_fee_pct}" + ))); + } + // Initialize the probabilistic scorer with default parameters for learning from payment // history. These parameters control how much successful/failed payments affect routing // scores and how quickly these scores decay over time. @@ -551,6 +590,7 @@ impl SimNode { pathfinding_graph, scorer: Mutex::new(scorer), clock, + max_route_fee_pct, }) } @@ -616,12 +656,20 @@ fn node_info(pubkey: PublicKey, alias: String) -> NodeInfo { } } -/// Uses LDK's pathfinding algorithm with default parameters to find a path from source to destination, with no -/// restrictions on fee budget. +/// The default maximum total routing fee that a payment will pay, expressed as a percentage of the payment +/// amount. Real senders place a limit on the fees that they are prepared to pay; without one, pathfinding will +/// happily route through channels advertising absurd fees (which some datasets use to mark channel directions +/// as unusable), silently draining the sender's liquidity in fees. Five percent is a generous budget by real +/// node standards, while still refusing pathological routes. +pub const DEFAULT_MAX_ROUTE_FEE_PCT: f64 = 5.0; + +/// Uses LDK's pathfinding algorithm with default parameters to find a path from source to destination, with the +/// total routing fees for the payment limited to the budget provided. async fn find_payment_route( source: &PublicKey, dest: PublicKey, amount_msat: u64, + max_fee_msat: u64, pathfinding_graph: &LdkNetworkGraph, scorer: &Mutex, Arc>>, ) -> Result { @@ -636,7 +684,7 @@ async fn find_payment_route( // Allow sending htlcs up to 50% of the channel's capacity. .with_max_channel_saturation_power_of_half(1), final_value_msat: amount_msat, - max_total_routing_fee_msat: None, + max_total_routing_fee_msat: Some(max_fee_msat), }, pathfinding_graph, None, @@ -683,10 +731,12 @@ impl LightningNode for SimNode { }; // Use the stored scorer when finding a route + let max_fee_msat = (amount_msat as f64 * self.max_route_fee_pct / 100.0) as u64; let route = match find_payment_route( &self.info.pubkey, dest, amount_msat, + max_fee_msat, &self.pathfinding_graph, &self.scorer, ) @@ -1023,6 +1073,49 @@ async fn handle_intercepted_htlc( Ok(Ok(attached_custom_records)) } +/// Configuration for optional "teleport" rebalancing of simulated channels, which periodically resets channels +/// whose available liquidity has drawn down. See [`SimGraph::start_rebalancer`]. +#[derive(Clone, Debug)] +pub struct RebalanceConfig { + /// The fraction of a channel's total capacity beneath which a side's available balance triggers a rebalance. + trigger_below_fraction: f64, + /// The time between scans of the network's channels. + interval: Duration, + /// Nodes that do not participate in rebalancing. A channel is skipped by the rebalancer if either of its + /// peers is excluded, regardless of which side of the channel has depleted. + exclude: HashSet, +} + +impl RebalanceConfig { + /// Creates a new rebalancing configuration, validating that the trigger fraction is in (0, 0.5) and that the + /// scan interval is non-zero. The trigger must be beneath 0.5 because both sides of a channel cannot be at or + /// above half of its capacity, so any higher value would rebalance every channel on every scan. Channels that + /// have a peer in `exclude` will never be rebalanced. + pub fn new( + trigger_below_fraction: f64, + interval: Duration, + exclude: HashSet, + ) -> Result { + if !(trigger_below_fraction > 0.0 && trigger_below_fraction < 0.5) { + return Err(SimulationError::SimulatedNetworkError(format!( + "rebalance trigger fraction must be in (0, 0.5), got: {trigger_below_fraction}" + ))); + } + + if interval.is_zero() { + return Err(SimulationError::SimulatedNetworkError( + "rebalance interval must be non-zero".to_string(), + )); + } + + Ok(RebalanceConfig { + trigger_below_fraction, + interval, + exclude, + }) + } +} + /// Graph is the top level struct that is used to coordinate simulation of lightning nodes. pub struct SimGraph { /// nodes caches the list of nodes in the network with a vector of their channel ids, only used for quick @@ -1099,13 +1192,78 @@ impl SimGraph { shutdown_signal, }) } + + /// Spawns a task that periodically "teleport" rebalances simulated channels whose available liquidity has + /// drawn down beneath the configured trigger, resetting them to an even split of their settled balance. This + /// is an opt-in escape hatch for long-running simulations: without any restoring force, random payment + /// activity progressively drains channels to one side and payment success rates decay, whereas real node + /// operators counteract depletion with out-of-band actions (circular rebalances, swaps, splices) that a + /// payment-only simulator cannot express directly. + /// + /// The task introduces no new source of randomness to the simulation: scans happen at a fixed interval on + /// the simulation clock, channels are visited in stable short channel ID order and the rebalance itself + /// involves no randomness. Each rebalance is logged with the amount moved, so the total volume of + /// intervention needed to keep the network liquid is observable in the simulation's logs. + /// + /// Nodes in the config's exclusion list do not participate in rebalancing: any channel that they are a peer + /// on is skipped by scans, no matter which side of the channel has depleted. This models operators that do + /// not maintain their liquidity, and allows their channels to drain naturally while the rest of the network + /// is kept liquid. + /// + /// The task runs until the simulation's shutdown signal is triggered. + pub fn start_rebalancer(&self, clock: Arc, cfg: RebalanceConfig) { + let channels = Arc::clone(&self.channels); + let listener = self.shutdown_signal.1.clone(); + + self.tasks.spawn(async move { + log::info!( + "Started channel rebalancer: triggering below {} of channel capacity, scanning every {:?}, {} nodes excluded.", + cfg.trigger_below_fraction, + cfg.interval, + cfg.exclude.len(), + ); + + loop { + select! { + _ = listener.clone() => { + log::debug!("Channel rebalancer received shutdown signal."); + return; + }, + _ = clock.sleep(cfg.interval) => {}, + } + + let mut channels_lock = channels.lock().await; + let mut scids: Vec = channels_lock.keys().copied().collect(); + scids.sort_unstable(); + + for scid in scids { + // Unwrap is safe because the key was drawn from the map and the lock is held throughout. + let channel = channels_lock.get_mut(&scid).unwrap(); + + // Channels that have an excluded peer never rebalance, regardless of which side of the + // channel has drawn down. + if cfg.exclude.contains(&channel.node_1.policy.pubkey) + || cfg.exclude.contains(&channel.node_2.policy.pubkey) + { + continue; + } + if let Some(moved_msat) = channel.rebalance_if_depleted(cfg.trigger_below_fraction) + { + log::info!("Rebalanced channel: {scid}, moved {moved_msat} msat."); + } + } + } + }); + } } -/// Produces a map of node public key to lightning node implementation to be used for simulations. +/// Produces a map of node public key to lightning node implementation to be used for simulations. Nodes will not +/// pay more than `max_route_fee_pct` percent of a payment's amount in routing fees. pub async fn ln_node_from_graph( graph: Arc>, routing_graph: Arc, clock: Arc, + max_route_fee_pct: f64, ) -> Result>>>, LightningError> { let sim_graph = graph.lock().await; let mut nodes: HashMap>>> = @@ -1119,6 +1277,7 @@ pub async fn ln_node_from_graph( graph.clone(), routing_graph.clone(), clock.clone(), + max_route_fee_pct, )?)), ); } @@ -1711,6 +1870,180 @@ mod tests { }; } + /// Tests validation of rebalancing configurations. + #[test] + fn test_rebalance_config_validation() { + assert!(RebalanceConfig::new(0.1, Duration::from_secs(600), HashSet::new()).is_ok()); + assert!(RebalanceConfig::new(0.0, Duration::from_secs(600), HashSet::new()).is_err()); + assert!(RebalanceConfig::new(0.5, Duration::from_secs(600), HashSet::new()).is_err()); + assert!(RebalanceConfig::new(-0.1, Duration::from_secs(600), HashSet::new()).is_err()); + assert!(RebalanceConfig::new(0.1, Duration::ZERO, HashSet::new()).is_err()); + } + + /// Tests that the rebalancing task resets depleted channels on its scan interval, and leaves healthy channels + /// untouched. Runs on a sped up simulation clock so that scan intervals elapse quickly in real time. + #[tokio::test] + async fn test_start_rebalancer() { + let capacity = 100_000_000; + let balanced = SimulatedChannel { + capacity_msat: capacity, + short_channel_id: ShortChannelID::from(0), + node_1: ChannelState::new(create_test_policy(capacity / 2), capacity / 2), + node_2: ChannelState::new(create_test_policy(capacity / 2), capacity / 2), + exclude_capacity: false, + }; + let depleted = SimulatedChannel { + capacity_msat: capacity, + short_channel_id: ShortChannelID::from(1), + node_1: ChannelState::new(create_test_policy(capacity / 2), capacity / 50), + node_2: ChannelState::new(create_test_policy(capacity / 2), capacity - capacity / 50), + exclude_capacity: false, + }; + // A channel that is just as depleted, but has a peer that is excluded from rebalancing. We exclude the + // node on the channel's healthy side to demonstrate that exclusion applies to the whole channel, not + // just to the side that has drawn down. + let excluded_depleted = SimulatedChannel { + capacity_msat: capacity, + short_channel_id: ShortChannelID::from(2), + node_1: ChannelState::new(create_test_policy(capacity / 2), capacity / 50), + node_2: ChannelState::new(create_test_policy(capacity / 2), capacity - capacity / 50), + exclude_capacity: false, + }; + let excluded_pubkey = excluded_depleted.node_2.policy.pubkey; + + let (shutdown_trigger, shutdown_listener) = trigger(); + let tasks = TaskTracker::new(); + let graph = SimGraph::new( + vec![balanced, depleted, excluded_depleted], + tasks.clone(), + vec![], + CustomRecords::default(), + (shutdown_trigger.clone(), shutdown_listener), + ) + .unwrap(); + + let clock = Arc::new(SimulationClock::new(1000).unwrap()); + graph.start_rebalancer( + clock, + RebalanceConfig::new( + 0.1, + Duration::from_secs(60), + HashSet::from([excluded_pubkey]), + ) + .unwrap(), + ); + + // The clock speeds wall time up by 1000x, so the rebalancer's 60 second scan interval elapses after 60ms + // of real time. Poll (with a generous timeout) until the depleted channel has been reset to an even split. + timeout(Duration::from_secs(5), async { + loop { + { + let channels = graph.channels.lock().await; + let depleted_channel = channels.get(&ShortChannelID::from(1)).unwrap(); + if depleted_channel.node_1.local_balance_msat == capacity / 2 { + assert_eq!(depleted_channel.node_2.local_balance_msat, capacity / 2); + depleted_channel.sanity_check().unwrap(); + + // The balanced channel should have been left untouched by the scans that have run. + let balanced_channel = channels.get(&ShortChannelID::from(0)).unwrap(); + assert_eq!(balanced_channel.node_1.local_balance_msat, capacity / 2); + assert_eq!(balanced_channel.node_2.local_balance_msat, capacity / 2); + break; + } + } + time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("depleted channel should be rebalanced within the timeout"); + + // Give the rebalancer a few more scan intervals, then check that the channel with an excluded peer has + // been skipped despite being just as depleted as the channel that was reset. + time::sleep(Duration::from_millis(300)).await; + { + let channels = graph.channels.lock().await; + let excluded_channel = channels.get(&ShortChannelID::from(2)).unwrap(); + assert_eq!(excluded_channel.node_1.local_balance_msat, capacity / 50); + assert_eq!( + excluded_channel.node_2.local_balance_msat, + capacity - capacity / 50 + ); + } + + // The rebalancer should exit cleanly on shutdown. + shutdown_trigger.trigger(); + tasks.close(); + tasks.wait().await; + } + + /// Tests rebalancing of channels whose liquidity has drawn down below a trigger fraction of capacity. + #[test] + fn test_rebalance_if_depleted() { + let capacity = 100_000_000; + let mut channel = SimulatedChannel { + capacity_msat: capacity, + short_channel_id: ShortChannelID::from(0), + node_1: ChannelState::new(create_test_policy(capacity / 2), capacity / 2), + node_2: ChannelState::new(create_test_policy(capacity / 2), capacity / 2), + exclude_capacity: false, + }; + let node_1_pk = channel.node_1.policy.pubkey; + + // A balanced channel is left untouched. + assert!(channel.rebalance_if_depleted(0.2).is_none()); + assert_channel_balances!(channel.node_1, capacity / 2, 0, 0); + assert_channel_balances!(channel.node_2, capacity / 2, 0, 0); + + // Draw node_1's side of the channel down beneath the trigger threshold and check that the channel is + // reset to an even split of its capacity. + channel.node_1.local_balance_msat = capacity / 25; + channel.node_2.local_balance_msat = capacity - capacity / 25; + + assert_eq!( + channel.rebalance_if_depleted(0.2), + Some(capacity / 2 - capacity / 25) + ); + assert_channel_balances!(channel.node_1, capacity / 2, 0, 0); + assert_channel_balances!(channel.node_2, capacity / 2, 0, 0); + channel.sanity_check().unwrap(); + + // With an HTLC in flight from node_1, only the settled portion of the channel's capacity is + // redistributed; the in-flight balance remains locked in the HTLC. + let htlc_amt = 1_000_000; + let hash = PaymentHash([1; 32]); + channel + .add_htlc( + &node_1_pk, + hash, + Htlc { + amount_msat: htlc_amt, + cltv_expiry: 40, + }, + ) + .unwrap() + .unwrap(); + + // Manually drain the remainder of node_1's settled balance over to node_2. + channel.node_2.local_balance_msat += channel.node_1.local_balance_msat - capacity / 25; + channel.node_1.local_balance_msat = capacity / 25; + channel.sanity_check().unwrap(); + + let settled = capacity - htlc_amt; + assert_eq!( + channel.rebalance_if_depleted(0.2), + Some(settled / 2 - capacity / 25) + ); + assert_channel_balances!(channel.node_1, settled / 2, 1, htlc_amt); + assert_channel_balances!(channel.node_2, settled - settled / 2, 0, 0); + channel.sanity_check().unwrap(); + + // Settling the in-flight HTLC after a rebalance keeps the channel's accounting consistent (remove_htlc + // runs the channel's sanity check internally). + channel.remove_htlc(&node_1_pk, &hash, true).unwrap(); + assert_channel_balances!(channel.node_1, settled / 2, 0, 0); + assert_channel_balances!(channel.node_2, settled - settled / 2 + htlc_amt, 0, 0); + } + /// Tests state updates related to adding and removing HTLCs to a channel. #[test] fn test_channel_state_transitions() { @@ -1984,7 +2317,7 @@ mod tests { let clock = Arc::new(SimulationClock::new(1).unwrap()); let routing_graph = Arc::new(populate_network_graph(channels, Arc::clone(&clock)).unwrap()); - let nodes = ln_node_from_graph(sim_graph, routing_graph, clock) + let nodes = ln_node_from_graph(sim_graph, routing_graph, clock, DEFAULT_MAX_ROUTE_FEE_PCT) .await .unwrap(); @@ -2092,6 +2425,66 @@ mod tests { } } + /// Tests that pathfinding limits total routing fees to the budget provided. Real senders place a limit on + /// the fees that they are prepared to pay, so a path that charges more than the budget should not be used + /// even if it is the only route to the destination (otherwise a single high-fee channel can drain the sender). + #[tokio::test] + async fn test_find_payment_route_fee_budget() { + let capacity = 10_000_000_000; + let mut channels = create_simulated_channels(2, capacity); + let source = channels[0].node_1.policy.pubkey; + let dest = channels[1].node_2.policy.pubkey; + let amount_msat = 1_000_000; + + let clock = Arc::new(SimulationClock::new(1).unwrap()); + + // With the modest fees that our test channels are created with, pathfinding should succeed with a fee + // budget of 5% of the payment amount. + let graph = Arc::new(populate_network_graph(channels.clone(), clock.clone()).unwrap()); + let scorer = Mutex::new(ProbabilisticScorer::new( + ProbabilisticScoringDecayParameters::default(), + graph.clone(), + Arc::new(WrappedLog {}), + )); + assert!(find_payment_route( + &source, + dest, + amount_msat, + amount_msat / 20, + &graph, + &scorer + ) + .await + .is_ok()); + + // Set the forwarding node's fee for the second channel to 10% of the payment amount. The only path to + // the destination now exceeds a 5% fee budget, so pathfinding should fail. + channels[1].node_1.policy.base_fee = amount_msat / 10; + let graph = Arc::new(populate_network_graph(channels, clock).unwrap()); + let scorer = Mutex::new(ProbabilisticScorer::new( + ProbabilisticScoringDecayParameters::default(), + graph.clone(), + Arc::new(WrappedLog {}), + )); + assert!(find_payment_route( + &source, + dest, + amount_msat, + amount_msat / 20, + &graph, + &scorer + ) + .await + .is_err()); + + // A more generous budget of 20% of the payment amount allows the payment to route again. + assert!( + find_payment_route(&source, dest, amount_msat, amount_msat / 5, &graph, &scorer) + .await + .is_ok() + ); + } + /// Tests the functionality of a `SimNode`, mocking out the `SimNetwork` that is responsible for payment /// propagation to isolate testing to just the implementation of `LightningNode`. #[tokio::test] @@ -2109,6 +2502,7 @@ mod tests { sim_network.clone(), Arc::new(graph), Arc::new(SystemClock {}), + DEFAULT_MAX_ROUTE_FEE_PCT, ) .unwrap(); @@ -2167,8 +2561,9 @@ mod tests { ); // Dispatch payments to different destinations and assert that our track payment results are as expected. - let hash_1 = node.send_payment(dest_1, 10_000).await.unwrap(); - let hash_2 = node.send_payment(dest_2, 15_000).await.unwrap(); + // Amounts are large enough that route fees fit within pathfinding's fee budget of 5% of the payment amount. + let hash_1 = node.send_payment(dest_1, 1_000_000).await.unwrap(); + let hash_2 = node.send_payment(dest_2, 1_500_000).await.unwrap(); let (_, shutdown_listener) = triggered::trigger(); @@ -2199,6 +2594,7 @@ mod tests { Arc::new(Mutex::new(test_kit.graph)), test_kit.routing_graph.clone(), Arc::new(SystemClock {}), + DEFAULT_MAX_ROUTE_FEE_PCT, ) .unwrap(); @@ -2359,9 +2755,16 @@ mod tests { dest: PublicKey, amt: u64, ) -> (Route, Result) { - let route = find_payment_route(&source, dest, amt, &self.routing_graph, &self.scorer) - .await - .unwrap(); + let route = find_payment_route( + &source, + dest, + amt, + amt / 20, + &self.routing_graph, + &self.scorer, + ) + .await + .unwrap(); let (sender, receiver) = oneshot::channel(); self.graph @@ -2394,8 +2797,9 @@ mod tests { let mut test_kit = DispatchPaymentTestKit::new(chan_capacity, vec![], CustomRecords::default()).await; - // Send a payment that should succeed from Alice -> Dave. - let mut amt = 20_000; + // Send a payment that should succeed from Alice -> Dave. The amount is large enough that the fees for the + // route comfortably fit within pathfinding's fee budget of 5% of the payment amount. + let mut amt = 1_000_000; let (route, _) = test_kit .send_test_payment(test_kit.nodes[0], test_kit.nodes[3], amt) .await; @@ -2444,7 +2848,7 @@ mod tests { // Finally, we'll test a multi-hop failure by trying to send from Alice -> Dave. Since Bob's liquidity is // drained, we expect a failure and unchanged balances along the route. let _ = test_kit - .send_test_payment(test_kit.nodes[0], test_kit.nodes[3], 20_000) + .send_test_payment(test_kit.nodes[0], test_kit.nodes[3], 1_000_000) .await; assert_eq!(test_kit.channel_balances().await, expected_balances); @@ -2460,8 +2864,9 @@ mod tests { let mut test_kit = DispatchPaymentTestKit::new(chan_capacity, vec![], CustomRecords::default()).await; - // Send a payment that should succeed from Alice -> Dave. - let amt = 20_000; + // Send a payment that should succeed from Alice -> Dave. The amount is large enough that the fees for the + // route comfortably fit within pathfinding's fee budget of 5% of the payment amount. + let amt = 1_000_000; let (route, _) = test_kit .send_test_payment(test_kit.nodes[0], test_kit.nodes[3], amt) .await; @@ -2570,6 +2975,7 @@ mod tests { Arc::new(Mutex::new(test_kit.graph)), test_kit.routing_graph.clone(), Arc::new(SystemClock {}), + DEFAULT_MAX_ROUTE_FEE_PCT, ) .unwrap();