From 412b1202f7031dbead9db20995eb977f055fe6c6 Mon Sep 17 00:00:00 2001 From: benthecarman Date: Thu, 9 Jul 2026 18:12:05 -0500 Subject: [PATCH 1/3] Update LDK dependencies Update LDK and payment instructions to compatible revisions and adapt forwarded and unified payment handling to their API changes. Keep both native and UniFFI test targets compiling. AI-assisted-by: OpenAI Codex --- Cargo.toml | 30 +++++++++++++++--------------- src/event.rs | 6 +++--- 2 files changed, 18 insertions(+), 18 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index c9ff50d22d..3cade1604d 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -41,18 +41,18 @@ postgres = ["dep:tokio-postgres", "dep:native-tls", "dep:postgres-native-tls"] #lightning-macros = { version = "0.2.0" } #lightning-dns-resolver = { version = "0.3.0" } -lightning = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "506cb91f2e0fb87906188b79777bcf42595d3623", features = ["std"] } -lightning-types = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "506cb91f2e0fb87906188b79777bcf42595d3623" } -lightning-invoice = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "506cb91f2e0fb87906188b79777bcf42595d3623", features = ["std"] } -lightning-net-tokio = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "506cb91f2e0fb87906188b79777bcf42595d3623" } -lightning-persister = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "506cb91f2e0fb87906188b79777bcf42595d3623", features = ["tokio"] } -lightning-background-processor = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "506cb91f2e0fb87906188b79777bcf42595d3623" } -lightning-rapid-gossip-sync = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "506cb91f2e0fb87906188b79777bcf42595d3623" } -lightning-block-sync = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "506cb91f2e0fb87906188b79777bcf42595d3623", features = ["rest-client", "rpc-client", "tokio"] } -lightning-transaction-sync = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "506cb91f2e0fb87906188b79777bcf42595d3623", features = ["esplora-async-https", "time", "electrum-rustls-ring"] } -lightning-liquidity = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "506cb91f2e0fb87906188b79777bcf42595d3623", features = ["std"] } -lightning-macros = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "506cb91f2e0fb87906188b79777bcf42595d3623" } -lightning-dns-resolver = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "506cb91f2e0fb87906188b79777bcf42595d3623" } +lightning = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "f680fd37d8bea69383e1b18730423f347e6a8341", features = ["std"] } +lightning-types = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "f680fd37d8bea69383e1b18730423f347e6a8341" } +lightning-invoice = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "f680fd37d8bea69383e1b18730423f347e6a8341", features = ["std"] } +lightning-net-tokio = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "f680fd37d8bea69383e1b18730423f347e6a8341" } +lightning-persister = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "f680fd37d8bea69383e1b18730423f347e6a8341", features = ["tokio"] } +lightning-background-processor = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "f680fd37d8bea69383e1b18730423f347e6a8341" } +lightning-rapid-gossip-sync = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "f680fd37d8bea69383e1b18730423f347e6a8341" } +lightning-block-sync = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "f680fd37d8bea69383e1b18730423f347e6a8341", features = ["rest-client", "rpc-client", "tokio"] } +lightning-transaction-sync = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "f680fd37d8bea69383e1b18730423f347e6a8341", features = ["esplora-async-https", "time", "electrum-rustls-ring"] } +lightning-liquidity = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "f680fd37d8bea69383e1b18730423f347e6a8341", features = ["std"] } +lightning-macros = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "f680fd37d8bea69383e1b18730423f347e6a8341" } +lightning-dns-resolver = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "f680fd37d8bea69383e1b18730423f347e6a8341" } bdk_chain = { version = "0.23.3", default-features = false, features = ["std"] } bdk_esplora = { version = "0.22.2", default-features = false, features = ["async-https-rustls", "tokio"]} @@ -85,14 +85,14 @@ postgres-native-tls = { version = "0.5", default-features = false, features = [" vss-client = { package = "vss-client-ng", version = "0.6" } prost = { version = "0.11.6", default-features = false} #bitcoin-payment-instructions = { version = "0.6" } -bitcoin-payment-instructions = { git = "https://github.com/tnull/bitcoin-payment-instructions", rev = "0e430be98c09540624a68a68022ee0551e86d1be" } +bitcoin-payment-instructions = { git = "https://github.com/benthecarman/bitcoin-payment-instructions", rev = "766f554ee0e44f4fb209f14c56c8b8bd7dadb46b" } [target.'cfg(windows)'.dependencies] winapi = { version = "0.3", features = ["winbase"] } [dev-dependencies] -lightning = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "506cb91f2e0fb87906188b79777bcf42595d3623", features = ["std", "_test_utils"] } -lightning-persister = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "506cb91f2e0fb87906188b79777bcf42595d3623", features = ["tokio"] } +lightning = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "f680fd37d8bea69383e1b18730423f347e6a8341", features = ["std", "_test_utils"] } +lightning-persister = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "f680fd37d8bea69383e1b18730423f347e6a8341", features = ["tokio"] } rand = { version = "0.9.2", default-features = false, features = ["std", "thread_rng", "os_rng"] } proptest = "1.0.0" regex = "1.5.6" diff --git a/src/event.rs b/src/event.rs index 91ab7b27de..2a9d21a1b8 100644 --- a/src/event.rs +++ b/src/event.rs @@ -1451,7 +1451,7 @@ where from_prev_str, next_htlcs.len(), to_next_str, - outbound_amount_forwarded_msat.unwrap_or(0), + outbound_amount_forwarded_msat, fee_earned, ); } else { @@ -1462,7 +1462,7 @@ where from_prev_str, next_htlcs.len(), to_next_str, - outbound_amount_forwarded_msat.unwrap_or(0), + outbound_amount_forwarded_msat, fee_earned, ); } @@ -1496,7 +1496,7 @@ where total_fee_earned_msat, skimmed_fee_msat, claim_from_onchain_tx, - outbound_amount_forwarded_msat, + outbound_amount_forwarded_msat: Some(outbound_amount_forwarded_msat), }; self.event_queue.add_event(event).await.map_err(|e| { log_error!(self.logger, "Failed to push to event queue: {}", e); From 999db7ca04abd4815e641efdb43c664ca19fcdda Mon Sep 17 00:00:00 2001 From: benthecarman Date: Thu, 9 Jul 2026 22:13:43 -0500 Subject: [PATCH 2/3] Prefactor: Generalize storable object keys Rename the storage-key encoding method so implementations are no longer limited to hexadecimal output. This prepares the data store for compact keys used by larger identifiers. AI-assisted-by: OpenAI Codex --- src/data_store.rs | 10 +++++----- src/payment/store.rs | 2 +- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/src/data_store.rs b/src/data_store.rs index b1ed816df9..7505fb4e0f 100644 --- a/src/data_store.rs +++ b/src/data_store.rs @@ -26,7 +26,7 @@ pub(crate) trait StorableObject: Clone + Readable + Writeable { } pub(crate) trait StorableObjectId: std::hash::Hash + PartialEq + Eq { - fn encode_to_hex_str(&self) -> String; + fn encode_to_key(&self) -> String; } pub(crate) trait StorableObjectUpdate { @@ -115,7 +115,7 @@ where let _guard = self.mutation_lock.lock().await; let should_remove = { self.objects.lock().expect("lock").contains_key(id) }; if should_remove { - let store_key = id.encode_to_hex_str(); + let store_key = id.encode_to_key(); KVStore::remove( &*self.kv_store, &self.primary_namespace, @@ -185,7 +185,7 @@ where } fn encode_object(object: &SO) -> (String, Vec) { - (object.id().encode_to_hex_str(), object.encode()) + (object.id().encode_to_key(), object.encode()) } async fn persist_encoded(&self, store_key: String, data: Vec) -> Result<(), Error> { @@ -238,7 +238,7 @@ mod tests { } impl StorableObjectId for TestObjectId { - fn encode_to_hex_str(&self) -> String { + fn encode_to_key(&self) -> String { hex_utils::to_string(&self.id) } } @@ -354,7 +354,7 @@ mod tests { let id = TestObjectId { id: [42u8; 4] }; assert!(data_store.get(&id).is_none()); - let store_key = id.encode_to_hex_str(); + let store_key = id.encode_to_key(); // Check we start empty. assert!(KVStore::read(&*store, &primary_namespace, &secondary_namespace, &store_key) diff --git a/src/payment/store.rs b/src/payment/store.rs index 38583dd7e7..9abf8d7422 100644 --- a/src/payment/store.rs +++ b/src/payment/store.rs @@ -149,7 +149,7 @@ impl Readable for PaymentDetails { } impl StorableObjectId for PaymentId { - fn encode_to_hex_str(&self) -> String { + fn encode_to_key(&self) -> String { hex_utils::to_string(&self.0) } } From 81249d08013c28c3602162d8fc6adb12905f8d8e Mon Sep 17 00:00:00 2001 From: benthecarman Date: Fri, 10 Jul 2026 05:40:25 -0500 Subject: [PATCH 3/3] Add forwarded payment tracking Store individual forwarding events and aggregate them into channel and channel-pair statistics for fee and profitability tracking. Infer channel-pair allocations for multi-HTLC forwards by matching incoming and outgoing amounts in FIFO order. AI-assisted-by: OpenAI Codex --- bindings/ldk_node.udl | 26 + src/builder.rs | 174 +++++- src/config.rs | 140 ++++- src/data_store.rs | 116 +++- src/event.rs | 889 +++++++++++++++++++++++++++- src/ffi/types.rs | 46 +- src/io/mod.rs | 19 + src/io/utils.rs | 67 ++- src/lib.rs | 104 +++- src/payment/forwarding.rs | 200 +++++++ src/payment/mod.rs | 10 +- src/payment/store.rs | 1181 ++++++++++++++++++++++++++++++++++++- src/types.rs | 7 + 13 files changed, 2938 insertions(+), 41 deletions(-) create mode 100644 src/payment/forwarding.rs diff --git a/bindings/ldk_node.udl b/bindings/ldk_node.udl index d7e9a774f9..db193505de 100644 --- a/bindings/ldk_node.udl +++ b/bindings/ldk_node.udl @@ -11,6 +11,8 @@ typedef dictionary ElectrumSyncConfig; typedef dictionary TorConfig; +typedef enum ForwardedPaymentTrackingMode; + typedef interface NodeEntropy; typedef interface ProbingConfig; @@ -113,6 +115,7 @@ interface Node { OnchainPayment onchain_payment(); UnifiedPayment unified_payment(); Liquidity liquidity(); + Forwarding forwarding(); [Throws=NodeError] void lnurl_auth(string lnurl); [Throws=NodeError] @@ -184,6 +187,8 @@ typedef interface UnifiedPayment; typedef interface Liquidity; +typedef interface Forwarding; + [Error] enum NodeError { "AlreadyRunning", @@ -394,6 +399,9 @@ typedef string OfferId; [Custom] typedef string PaymentId; +[Custom] +typedef string ForwardedPaymentId; + [Custom] typedef string PaymentHash; @@ -406,6 +414,12 @@ typedef string PaymentSecret; [Custom] typedef string ChannelId; +[Custom] +typedef string ChannelPairStatsId; + +[Custom] +typedef string PageToken; + [Custom] typedef string UserChannelId; @@ -432,3 +446,15 @@ typedef enum Event; typedef interface HRNResolverConfig; typedef dictionary HumanReadableNamesConfig; + +typedef dictionary ForwardedPaymentDetails; + +typedef dictionary ChannelForwardingStats; + +typedef dictionary ChannelPairForwardingStats; + +typedef dictionary ForwardedPaymentDetailsPage; + +typedef dictionary ChannelForwardingStatsPage; + +typedef dictionary ChannelPairForwardingStatsPage; diff --git a/src/builder.rs b/src/builder.rs index 639838ff3a..4c26391d97 100644 --- a/src/builder.rs +++ b/src/builder.rs @@ -49,8 +49,9 @@ use crate::chain::ChainSource; use crate::config::{ default_user_config, may_announce_channel, AnnounceError, AsyncPaymentsRole, BitcoindRestClientConfig, Config, ElectrumSyncConfig, EsploraSyncConfig, HRNResolverConfig, - TorConfig, DEFAULT_ESPLORA_SERVER_URL, DEFAULT_LOG_FILENAME, DEFAULT_LOG_LEVEL, - DEFAULT_MAX_PROBE_AMOUNT_MSAT, DEFAULT_MIN_PROBE_AMOUNT_MSAT, + ResolvedForwardedPaymentBucketConfig, TorConfig, DEFAULT_ESPLORA_SERVER_URL, + DEFAULT_LOG_FILENAME, DEFAULT_LOG_LEVEL, DEFAULT_MAX_PROBE_AMOUNT_MSAT, + DEFAULT_MIN_PROBE_AMOUNT_MSAT, }; use crate::connection::ConnectionManager; use crate::entropy::NodeEntropy; @@ -60,12 +61,19 @@ use crate::gossip::GossipSource; use crate::io::sqlite_store::SqliteStore; use crate::io::utils::{ open_or_migrate_fs_store, read_all_objects, read_event_queue, - read_external_pathfinding_scores_from_cache, read_network_graph, read_node_metrics, - read_output_sweeper, read_peer_info, read_scorer, + read_external_pathfinding_scores_from_cache, read_forwarded_payment_bucket_size, + read_network_graph, read_node_metrics, read_output_sweeper, read_peer_info, read_scorer, + write_forwarded_payment_bucket_size, }; use crate::io::vss_store::VssStoreBuilder; use crate::io::{ - self, PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, + self, CHANNEL_FORWARDING_STATS_PERSISTENCE_PRIMARY_NAMESPACE, + CHANNEL_FORWARDING_STATS_PERSISTENCE_SECONDARY_NAMESPACE, + CHANNEL_PAIR_FORWARDING_STATS_PERSISTENCE_PRIMARY_NAMESPACE, + CHANNEL_PAIR_FORWARDING_STATS_PERSISTENCE_SECONDARY_NAMESPACE, + FORWARDED_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, + FORWARDED_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, + PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, PENDING_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, }; @@ -82,7 +90,8 @@ use crate::probing::{ use crate::runtime::{Runtime, RuntimeSpawner}; use crate::tx_broadcaster::TransactionBroadcaster; use crate::types::{ - AsyncPersister, ChainMonitor, ChannelManager, DynStore, DynStoreRef, DynStoreWrapper, + AsyncPersister, ChainMonitor, ChannelForwardingStatsStore, ChannelManager, + ChannelPairForwardingStatsStore, DynStore, DynStoreRef, DynStoreWrapper, ForwardedPaymentStore, GossipSync, Graph, HRNResolver, KeysManager, MessageRouter, OnionMessenger, PaymentStore, PeerManager, PendingPaymentStore, }; @@ -174,6 +183,10 @@ pub enum BuildError { InvalidTorProxyAddress, /// The provided alias is invalid. InvalidNodeAlias, + /// The forwarded-payment bucket length is zero or too large to represent in seconds. + InvalidForwardedPaymentBucketLength, + /// The forwarded-payment bucket length differs from the value previously persisted. + ForwardedPaymentBucketLengthMismatch, /// An attempt to setup a runtime has failed. RuntimeSetupFailed, /// We failed to read data from the [`KVStore`]. @@ -236,6 +249,12 @@ impl fmt::Display for BuildError { Self::LoggerSetupFailed => write!(f, "Failed to setup the logger."), Self::ChainSourceSetupFailed => write!(f, "Failed to setup the chain source."), Self::InvalidNodeAlias => write!(f, "Given node alias is invalid."), + Self::InvalidForwardedPaymentBucketLength => { + write!(f, "Forwarded-payment bucket length is invalid.") + }, + Self::ForwardedPaymentBucketLengthMismatch => { + write!(f, "Forwarded-payment bucket length does not match persisted data.") + }, Self::NetworkMismatch => { write!(f, "Given network does not match the node's previously configured network.") }, @@ -1409,6 +1428,12 @@ fn build_with_store_internal( ) -> Result { optionally_install_rustls_cryptoprovider(); + let configured_forwarded_payment_bucket_size_secs = + config.forwarded_payment_tracking_mode.bucket_size_secs().map_err(|_| { + log_error!(logger, "Forwarded-payment bucket length is zero or too large"); + BuildError::InvalidForwardedPaymentBucketLength + })?; + if let Err(err) = may_announce_channel(&config) { if config.announcement_addresses.is_some() { log_error!(logger, "Announcement addresses were set but some required configuration options for node announcement are missing: {}", err); @@ -1440,24 +1465,72 @@ fn build_with_store_internal( let kv_store_ref = Arc::clone(&kv_store); let logger_ref = Arc::clone(&logger); - let (payment_store_res, node_metris_res, pending_payment_store_res) = - runtime.block_on(async move { - tokio::join!( - read_all_objects( - &*kv_store_ref, - PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, - PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, - Arc::clone(&logger_ref), - ), - read_node_metrics(&*kv_store_ref, Arc::clone(&logger_ref)), - read_all_objects( - &*kv_store_ref, - PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, - PENDING_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, - Arc::clone(&logger_ref), - ) + let ( + payment_store_res, + forwarded_payment_store_res, + channel_forwarding_stats_res, + channel_pair_forwarding_stats_res, + forwarded_payment_bucket_size_res, + node_metris_res, + pending_payment_store_res, + ) = runtime.block_on(async move { + tokio::join!( + read_all_objects( + &*kv_store_ref, + PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, + PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, + Arc::clone(&logger_ref), + ), + read_all_objects( + &*kv_store_ref, + FORWARDED_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, + FORWARDED_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, + Arc::clone(&logger_ref), + ), + read_all_objects( + &*kv_store_ref, + CHANNEL_FORWARDING_STATS_PERSISTENCE_PRIMARY_NAMESPACE, + CHANNEL_FORWARDING_STATS_PERSISTENCE_SECONDARY_NAMESPACE, + Arc::clone(&logger_ref), + ), + read_all_objects( + &*kv_store_ref, + CHANNEL_PAIR_FORWARDING_STATS_PERSISTENCE_PRIMARY_NAMESPACE, + CHANNEL_PAIR_FORWARDING_STATS_PERSISTENCE_SECONDARY_NAMESPACE, + Arc::clone(&logger_ref), + ), + read_forwarded_payment_bucket_size(&*kv_store_ref, Arc::clone(&logger_ref)), + read_node_metrics(&*kv_store_ref, Arc::clone(&logger_ref)), + read_all_objects( + &*kv_store_ref, + PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, + PENDING_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, + Arc::clone(&logger_ref), ) - }); + ) + }); + + let persisted_forwarded_payment_bucket_size_secs = match forwarded_payment_bucket_size_res { + Ok(bucket_size_secs) => Some(bucket_size_secs), + Err(e) if e.kind() == std::io::ErrorKind::NotFound => None, + Err(e) => { + log_error!(logger, "Failed to read forwarded-payment bucket size: {}", e); + return Err(BuildError::ReadFailed); + }, + }; + let forwarded_payment_bucket_config = ResolvedForwardedPaymentBucketConfig::resolve( + configured_forwarded_payment_bucket_size_secs, + persisted_forwarded_payment_bucket_size_secs, + ) + .map_err(|_| { + log_error!( + logger, + "Configured forwarded-payment bucket size {:?} does not match persisted size {:?}", + configured_forwarded_payment_bucket_size_secs, + persisted_forwarded_payment_bucket_size_secs + ); + BuildError::ForwardedPaymentBucketLengthMismatch + })?; // Initialize the status fields. let node_metrics = match node_metris_res { @@ -1486,6 +1559,48 @@ fn build_with_store_internal( }, }; + let forwarded_payment_store = match forwarded_payment_store_res { + Ok(forwarded_payments) => Arc::new(ForwardedPaymentStore::new( + forwarded_payments, + FORWARDED_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE.to_string(), + FORWARDED_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE.to_string(), + Arc::clone(&kv_store), + Arc::clone(&logger), + )), + Err(e) => { + log_error!(logger, "Failed to read forwarded payment data from store: {}", e); + return Err(BuildError::ReadFailed); + }, + }; + + let channel_forwarding_stats_store = match channel_forwarding_stats_res { + Ok(stats) => Arc::new(ChannelForwardingStatsStore::new( + stats, + CHANNEL_FORWARDING_STATS_PERSISTENCE_PRIMARY_NAMESPACE.to_string(), + CHANNEL_FORWARDING_STATS_PERSISTENCE_SECONDARY_NAMESPACE.to_string(), + Arc::clone(&kv_store), + Arc::clone(&logger), + )), + Err(e) => { + log_error!(logger, "Failed to read channel forwarding stats from store: {}", e); + return Err(BuildError::ReadFailed); + }, + }; + + let channel_pair_forwarding_stats_store = match channel_pair_forwarding_stats_res { + Ok(stats) => Arc::new(ChannelPairForwardingStatsStore::new( + stats, + CHANNEL_PAIR_FORWARDING_STATS_PERSISTENCE_PRIMARY_NAMESPACE.to_string(), + CHANNEL_PAIR_FORWARDING_STATS_PERSISTENCE_SECONDARY_NAMESPACE.to_string(), + Arc::clone(&kv_store), + Arc::clone(&logger), + )), + Err(e) => { + log_error!(logger, "Failed to read channel pair forwarding stats from store: {}", e); + return Err(BuildError::ReadFailed); + }, + }; + let (chain_source, chain_tip_opt) = match chain_data_source_config { Some(ChainDataSourceConfig::Esplora { server_url, headers, sync_config }) => { let sync_config = sync_config.unwrap_or(EsploraSyncConfig::default()); @@ -2307,6 +2422,15 @@ fn build_with_store_internal( }) }); + if let Some(bucket_size_secs) = forwarded_payment_bucket_config.new_bucket_size_secs() { + runtime + .block_on(write_forwarded_payment_bucket_size(&*kv_store, bucket_size_secs)) + .map_err(|e| { + log_error!(logger, "Failed to persist forwarded-payment bucket size: {}", e); + BuildError::WriteFailed + })?; + } + Ok(Node { runtime, stop_sender, @@ -2334,6 +2458,10 @@ fn build_with_store_internal( scorer, peer_store, payment_store, + forwarded_payment_store, + channel_forwarding_stats_store, + channel_pair_forwarding_stats_store, + forwarded_payment_bucket_size_secs: forwarded_payment_bucket_config.bucket_size_secs(), lnurl_auth, is_running, node_metrics, diff --git a/src/config.rs b/src/config.rs index f168df94ef..ec3d17b011 100644 --- a/src/config.rs +++ b/src/config.rs @@ -128,6 +128,84 @@ pub(crate) const HRN_RESOLUTION_TIMEOUT_SECS: u64 = 5; // The timeout after which we abort an LNURL-auth operation. pub(crate) const LNURL_AUTH_TIMEOUT_SECS: u64 = 15; +/// The mode used for tracking forwarded payments. +#[derive(Copy, Clone, Debug, PartialEq, Eq)] +#[cfg_attr(feature = "uniffi", derive(uniffi::Enum))] +pub enum ForwardedPaymentTrackingMode { + /// Store individual forwarded payments for the current and previous aggregation buckets. + /// + /// Payments from older buckets are aggregated into channel-pair statistics and removed. + Detailed { + /// The length of each aggregation bucket, in minutes. Must be non-zero. + /// + /// The first detailed configuration persists this value. Later detailed configurations for + /// the same node must use the same value so persisted statistics retain consistent bucket + /// geometry. + bucket_length_minutes: u64, + }, + /// Track new forwarded payments only as per-channel aggregate statistics. + /// + /// If detailed tracking was previously configured, its persisted bucket length is reused to + /// aggregate and remove remaining detailed records. + Stats, +} + +impl Default for ForwardedPaymentTrackingMode { + fn default() -> Self { + Self::Stats + } +} + +impl ForwardedPaymentTrackingMode { + pub(crate) fn bucket_size_secs(self) -> Result, ()> { + match self { + Self::Detailed { bucket_length_minutes } => bucket_length_minutes + .checked_mul(60) + .filter(|bucket_size_secs| *bucket_size_secs > 0) + .map(Some) + .ok_or(()), + Self::Stats => Ok(None), + } + } +} + +/// The effective forwarded-payment bucket configuration after reconciling user configuration with +/// persisted state. +#[derive(Debug, PartialEq, Eq)] +pub(crate) struct ResolvedForwardedPaymentBucketConfig { + bucket_size_secs: Option, + new_bucket_size_secs: Option, +} + +impl ResolvedForwardedPaymentBucketConfig { + pub(crate) fn resolve( + configured_bucket_size_secs: Option, persisted_bucket_size_secs: Option, + ) -> Result { + match (configured_bucket_size_secs, persisted_bucket_size_secs) { + (Some(configured), Some(persisted)) if configured != persisted => Err(()), + (Some(configured), None) => Ok(Self { + bucket_size_secs: Some(configured), + new_bucket_size_secs: Some(configured), + }), + (Some(configured), Some(_)) => { + Ok(Self { bucket_size_secs: Some(configured), new_bucket_size_secs: None }) + }, + (None, Some(persisted)) => { + Ok(Self { bucket_size_secs: Some(persisted), new_bucket_size_secs: None }) + }, + (None, None) => Ok(Self { bucket_size_secs: None, new_bucket_size_secs: None }), + } + } + + pub(crate) fn bucket_size_secs(&self) -> Option { + self.bucket_size_secs + } + + pub(crate) fn new_bucket_size_secs(&self) -> Option { + self.new_bucket_size_secs + } +} + #[derive(Debug, Clone)] #[cfg_attr(feature = "uniffi", derive(uniffi::Record))] /// Represents the configuration of an [`Node`] instance. @@ -147,9 +225,10 @@ pub(crate) const LNURL_AUTH_TIMEOUT_SECS: u64 = 15; /// | `route_parameters` | None | /// | `tor_config` | None | /// | `hrn_config` | HumanReadableNamesConfig::default() | +/// | `forwarded_payment_tracking_mode` | Stats | /// -/// See [`AnchorChannelsConfig`] and [`RouteParametersConfig`] for more information regarding their -/// respective default values. +/// See [`AnchorChannelsConfig`], [`RouteParametersConfig`], and +/// [`ForwardedPaymentTrackingMode`] for more information regarding their respective default values. /// /// [`Node`]: crate::Node pub struct Config { @@ -222,6 +301,8 @@ pub struct Config { /// /// [BIP 353]: https://github.com/bitcoin/bips/blob/master/bip-0353.mediawiki pub hrn_config: HumanReadableNamesConfig, + /// The mode used for tracking forwarded payments. + pub forwarded_payment_tracking_mode: ForwardedPaymentTrackingMode, } impl Default for Config { @@ -238,6 +319,7 @@ impl Default for Config { route_parameters: None, node_alias: None, hrn_config: HumanReadableNamesConfig::default(), + forwarded_payment_tracking_mode: ForwardedPaymentTrackingMode::default(), } } } @@ -770,10 +852,62 @@ mod tests { use super::{ clamp_full_scan_stop_gap, may_announce_channel, AnnounceError, Config, ElectrumSyncConfig, - EsploraSyncConfig, NodeAlias, SocketAddress, DEFAULT_FULL_SCAN_STOP_GAP, + EsploraSyncConfig, ForwardedPaymentTrackingMode, NodeAlias, + ResolvedForwardedPaymentBucketConfig, SocketAddress, DEFAULT_FULL_SCAN_STOP_GAP, MAX_FULL_SCAN_STOP_GAP, MIN_FULL_SCAN_STOP_GAP, }; + #[test] + fn resolves_forwarded_payment_bucket_size_across_mode_changes() { + assert_eq!( + ResolvedForwardedPaymentBucketConfig::resolve(None, None), + Ok(ResolvedForwardedPaymentBucketConfig { + bucket_size_secs: None, + new_bucket_size_secs: None, + }) + ); + assert_eq!( + ResolvedForwardedPaymentBucketConfig::resolve(Some(900), None), + Ok(ResolvedForwardedPaymentBucketConfig { + bucket_size_secs: Some(900), + new_bucket_size_secs: Some(900), + }) + ); + assert_eq!( + ResolvedForwardedPaymentBucketConfig::resolve(None, Some(900)), + Ok(ResolvedForwardedPaymentBucketConfig { + bucket_size_secs: Some(900), + new_bucket_size_secs: None, + }) + ); + assert_eq!( + ResolvedForwardedPaymentBucketConfig::resolve(Some(900), Some(900)), + Ok(ResolvedForwardedPaymentBucketConfig { + bucket_size_secs: Some(900), + new_bucket_size_secs: None, + }) + ); + assert_eq!(ResolvedForwardedPaymentBucketConfig::resolve(Some(1_800), Some(900)), Err(())); + } + + #[test] + fn forwarded_payment_bucket_length_must_be_nonzero_and_fit_seconds() { + assert_eq!(ForwardedPaymentTrackingMode::Stats.bucket_size_secs(), Ok(None)); + assert_eq!( + ForwardedPaymentTrackingMode::Detailed { bucket_length_minutes: 1 }.bucket_size_secs(), + Ok(Some(60)) + ); + assert_eq!( + ForwardedPaymentTrackingMode::Detailed { bucket_length_minutes: 0 }.bucket_size_secs(), + Err(()) + ); + assert_eq!( + ForwardedPaymentTrackingMode::Detailed { bucket_length_minutes: u64::MAX } + .bucket_size_secs(), + Err(()) + ); + } + #[test] fn node_announce_channel() { // Default configuration with node alias and listening addresses unset diff --git a/src/data_store.rs b/src/data_store.rs index 7505fb4e0f..b5bbf9bd4e 100644 --- a/src/data_store.rs +++ b/src/data_store.rs @@ -9,11 +9,11 @@ use std::collections::HashMap; use std::ops::Deref; use std::sync::{Arc, Mutex}; -use lightning::util::persist::KVStore; +use lightning::util::persist::{KVStore, PageToken, PaginatedKVStore}; use lightning::util::ser::{Readable, Writeable}; use crate::logger::{log_error, LdkLogger}; -use crate::types::DynStore; +use crate::types::{DynStore, DynStoreRef}; use crate::Error; pub(crate) trait StorableObject: Clone + Readable + Writeable { @@ -73,7 +73,18 @@ where } pub(crate) async fn insert(&self, object: SO) -> Result { + self.insert_with(|| object).await + } + + /// Builds and inserts an object while holding the mutation lock. + /// + /// This is useful when fields on the object, such as a timestamp, must be assigned atomically + /// with respect to other store mutations. + pub(crate) async fn insert_with SO>( + &self, build_object: F, + ) -> Result { let _guard = self.mutation_lock.lock().await; + let object = build_object(); self.persist(&object).await?; let mut locked_objects = self.objects.lock().expect("lock"); @@ -179,6 +190,72 @@ where self.objects.lock().expect("lock").values().filter(f).cloned().collect::>() } + pub(crate) async fn list_page( + &self, page_token: Option, + ) -> Result<(Vec, Option), Error> { + // Keep removals from invalidating keys between listing the page and reading its objects. + let _guard = self.mutation_lock.lock().await; + let response = PaginatedKVStore::list_paginated( + &DynStoreRef(Arc::clone(&self.kv_store)), + &self.primary_namespace, + &self.secondary_namespace, + page_token, + ) + .await + .map_err(|e| { + log_error!( + self.logger, + "Listing object data under {}/{} failed due to: {}", + &self.primary_namespace, + &self.secondary_namespace, + e + ); + Error::PersistenceFailed + })?; + + let mut objects = Vec::with_capacity(response.keys.len()); + for key in response.keys { + let data = KVStore::read( + &DynStoreRef(Arc::clone(&self.kv_store)), + &self.primary_namespace, + &self.secondary_namespace, + &key, + ) + .await + .map_err(|e| { + log_error!( + self.logger, + "Reading object data for key {}/{}/{} failed due to: {}", + &self.primary_namespace, + &self.secondary_namespace, + key, + e + ); + Error::PersistenceFailed + })?; + + let object = SO::read(&mut &data[..]).map_err(|e| { + log_error!( + self.logger, + "Failed to deserialize object data for key {}/{}/{}: {}", + &self.primary_namespace, + &self.secondary_namespace, + key, + e + ); + Error::PersistenceFailed + })?; + objects.push(object); + } + + Ok((objects, response.next_page_token)) + } + + /// Prevents mutations to this store until the returned guard is dropped. + pub(crate) async fn mutation_guard(&self) -> tokio::sync::MutexGuard<'_, ()> { + self.mutation_lock.lock().await + } + async fn persist(&self, object: &SO) -> Result<(), Error> { let (store_key, data) = Self::encode_object(object); self.persist_encoded(store_key, data).await @@ -223,6 +300,8 @@ where #[cfg(test)] mod tests { + use std::sync::atomic::{AtomicBool, Ordering}; + use lightning::util::persist::{PageToken, PaginatedKVStore, PaginatedListResponse}; use lightning::util::test_utils::TestLogger; use lightning::{impl_writeable_tlv_based, io}; @@ -403,6 +482,39 @@ mod tests { assert_eq!(Ok(true), data_store.insert_or_update(new_iou_object).await); } + #[tokio::test] + async fn insert_with_builds_object_while_mutations_are_locked() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let logger = Arc::new(TestLogger::new()); + let data_store: Arc>> = Arc::new(DataStore::new( + Vec::new(), + "datastore_test_primary".to_string(), + "datastore_test_secondary".to_string(), + store, + logger, + )); + let guard = data_store.mutation_guard().await; + let build_called = Arc::new(AtomicBool::new(false)); + let insert_handle = { + let data_store = Arc::clone(&data_store); + let build_called = Arc::clone(&build_called); + tokio::spawn(async move { + data_store + .insert_with(|| { + build_called.store(true, Ordering::Relaxed); + TestObject { id: TestObjectId { id: [42; 4] }, data: [23; 3] } + }) + .await + }) + }; + + tokio::task::yield_now().await; + assert!(!build_called.load(Ordering::Relaxed)); + drop(guard); + assert_eq!(insert_handle.await.unwrap(), Ok(false)); + assert!(build_called.load(Ordering::Relaxed)); + } + #[tokio::test] async fn insert_or_update_does_not_mutate_memory_if_persist_fails() { let existing_id = TestObjectId { id: [42u8; 4] }; diff --git a/src/event.rs b/src/event.rs index 2a9d21a1b8..ed08c67c19 100644 --- a/src/event.rs +++ b/src/event.rs @@ -7,9 +7,10 @@ use core::future::Future; use core::task::{Poll, Waker}; -use std::collections::VecDeque; +use std::collections::{HashMap, VecDeque}; use std::ops::Deref; use std::sync::{Arc, Mutex}; +use std::time::{SystemTime, UNIX_EPOCH}; use bitcoin::blockdata::locktime::absolute::LockTime; use bitcoin::secp256k1::PublicKey; @@ -34,7 +35,7 @@ use lightning::{impl_writeable_tlv_based, impl_writeable_tlv_based_enum}; use lightning_liquidity::lsps2::utils::compute_opening_fee; use lightning_types::payment::{PaymentHash, PaymentPreimage}; -use crate::config::{may_announce_channel, Config}; +use crate::config::{may_announce_channel, Config, ForwardedPaymentTrackingMode}; use crate::connection::ConnectionManager; use crate::data_store::DataStoreUpdateResult; use crate::fee_estimator::ConfirmationTarget; @@ -49,13 +50,15 @@ use crate::logger::{log_debug, log_error, log_info, log_trace, LdkLogger, Logger use crate::payment::asynchronous::om_mailbox::OnionMessageMailbox; use crate::payment::asynchronous::static_invoice_store::StaticInvoiceStore; use crate::payment::store::{ - PaymentDetails, PaymentDetailsUpdate, PaymentDirection, PaymentKind, PaymentStatus, + ChannelForwardingStats, ForwardedPaymentDetails, ForwardedPaymentId, PaymentDetails, + PaymentDetailsUpdate, PaymentDirection, PaymentKind, PaymentStatus, }; use crate::payment::PaymentMetadata; use crate::probing::Prober; use crate::runtime::Runtime; use crate::types::{ - CustomTlvRecord, DynStore, KeysManager, OnionMessenger, PaymentStore, Sweeper, Wallet, + ChannelForwardingStatsStore, CustomTlvRecord, DynStore, ForwardedPaymentStore, KeysManager, + OnionMessenger, PaymentStore, Sweeper, Wallet, }; use crate::{ hex_utils, BumpTransactionEventHandler, ChannelManager, Error, Graph, PeerInfo, PeerStore, @@ -68,6 +71,8 @@ use crate::{ pub struct HTLCLocator { /// The channel that the HTLC was sent or received on. pub channel_id: ChannelId, + /// The amount of the HTLC, in millisatoshis, if known. + pub amount_msat: Option, /// The `user_channel_id` for the channel. /// /// Will only be `None` for events serialized with LDK Node v0.3.0 or prior, or if the @@ -84,18 +89,187 @@ impl_writeable_tlv_based!(HTLCLocator, { (1, channel_id, required), (3, user_channel_id, option), (5, node_id, option), + (7, amount_msat, option), }); impl From for HTLCLocator { fn from(value: LdkHtlcLocator) -> Self { HTLCLocator { channel_id: value.channel_id, + amount_msat: value.amount_msat, user_channel_id: value.user_channel_id.map(|u| UserChannelId(u)), node_id: value.node_id, } } } +#[derive(Clone, Debug, PartialEq, Eq)] +struct ForwardedPaymentAllocation { + prev_htlc_index: usize, + next_htlc_index: usize, + inbound_amount_forwarded_msat: u64, + total_fee_earned_msat: Option, + skimmed_fee_msat: Option, + claim_from_onchain_tx: bool, + outbound_amount_forwarded_msat: u64, +} + +fn forwarded_payment_allocations( + prev_htlcs: &[LdkHtlcLocator], next_htlcs: &[LdkHtlcLocator], + total_fee_earned_msat: Option, skimmed_fee_msat: Option, claim_from_onchain_tx: bool, +) -> Option> { + if prev_htlcs.is_empty() || next_htlcs.is_empty() { + return None; + } + + let prev_amounts = + prev_htlcs.iter().map(|htlc| htlc.amount_msat).collect::>>()?; + let next_amounts = + next_htlcs.iter().map(|htlc| htlc.amount_msat).collect::>>()?; + + if prev_htlcs.len() == 1 && next_htlcs.len() == 1 { + return Some(vec![ForwardedPaymentAllocation { + prev_htlc_index: 0, + next_htlc_index: 0, + inbound_amount_forwarded_msat: prev_amounts[0], + total_fee_earned_msat, + skimmed_fee_msat, + claim_from_onchain_tx, + outbound_amount_forwarded_msat: next_amounts[0], + }]); + } + + // LDK reports the incoming and outgoing HTLCs, but not which incoming HTLC funded each + // outgoing HTLC. Infer the relationship by treating both lists as FIFO queues: allocate the + // smaller of the current HTLCs' remaining amounts, then advance past whichever HTLC is + // exhausted. If one side is exhausted first, attribute the other side's unmatched remainder to + // its final HTLC. This preserves every HTLC's exact amount without producing a Cartesian product + // of every possible channel pair. + let mut allocations = Vec::with_capacity(prev_htlcs.len() + next_htlcs.len()); + let mut prev_htlc_index = 0; + let mut next_htlc_index = 0; + let mut remaining_inbound_amount_msat = prev_amounts[0]; + let mut remaining_outbound_amount_msat = next_amounts[0]; + + while prev_htlc_index < prev_htlcs.len() && next_htlc_index < next_htlcs.len() { + if remaining_inbound_amount_msat == 0 { + prev_htlc_index += 1; + if prev_htlc_index < prev_htlcs.len() { + remaining_inbound_amount_msat = prev_amounts[prev_htlc_index]; + } + continue; + } + if remaining_outbound_amount_msat == 0 { + next_htlc_index += 1; + if next_htlc_index < next_htlcs.len() { + remaining_outbound_amount_msat = next_amounts[next_htlc_index]; + } + continue; + } + + let amount = remaining_inbound_amount_msat.min(remaining_outbound_amount_msat); + allocations.push(ForwardedPaymentAllocation { + prev_htlc_index, + next_htlc_index, + inbound_amount_forwarded_msat: amount, + total_fee_earned_msat: total_fee_earned_msat.map(|_| 0), + skimmed_fee_msat: skimmed_fee_msat.map(|_| 0), + claim_from_onchain_tx, + outbound_amount_forwarded_msat: amount, + }); + remaining_inbound_amount_msat -= amount; + remaining_outbound_amount_msat -= amount; + } + + // Any unmatched inbound amount represents value not present in the outgoing HTLCs, such as the + // forwarding fee. Attribute it to the final outgoing HTLC so each incoming HTLC's full amount + // remains represented. + let last_next_htlc_index = next_htlcs.len() - 1; + while prev_htlc_index < prev_htlcs.len() { + if remaining_inbound_amount_msat > 0 { + allocations.push(ForwardedPaymentAllocation { + prev_htlc_index, + next_htlc_index: last_next_htlc_index, + inbound_amount_forwarded_msat: remaining_inbound_amount_msat, + total_fee_earned_msat: total_fee_earned_msat.map(|_| 0), + skimmed_fee_msat: skimmed_fee_msat.map(|_| 0), + claim_from_onchain_tx, + outbound_amount_forwarded_msat: 0, + }); + } + prev_htlc_index += 1; + if prev_htlc_index < prev_htlcs.len() { + remaining_inbound_amount_msat = prev_amounts[prev_htlc_index]; + } + } + + // The inverse should not normally occur because forwarding fees make the inbound total larger, + // but preserve any unmatched outbound amount symmetrically if it does. + let last_prev_htlc_index = prev_htlcs.len() - 1; + while next_htlc_index < next_htlcs.len() { + if remaining_outbound_amount_msat > 0 { + allocations.push(ForwardedPaymentAllocation { + prev_htlc_index: last_prev_htlc_index, + next_htlc_index, + inbound_amount_forwarded_msat: 0, + total_fee_earned_msat: total_fee_earned_msat.map(|_| 0), + skimmed_fee_msat: skimmed_fee_msat.map(|_| 0), + claim_from_onchain_tx, + outbound_amount_forwarded_msat: remaining_outbound_amount_msat, + }); + } + next_htlc_index += 1; + if next_htlc_index < next_htlcs.len() { + remaining_outbound_amount_msat = next_amounts[next_htlc_index]; + } + } + + // LDK reports fees for the complete forward rather than for individual HTLCs. Store them on the + // final inferred allocation so summing the allocations recovers the event's totals without + // counting the fees more than once. + if let Some(total_fee_earned_msat) = total_fee_earned_msat { + allocations.last_mut()?.total_fee_earned_msat = Some(total_fee_earned_msat); + } + + if let Some(skimmed_fee_msat) = skimmed_fee_msat { + allocations.last_mut()?.skimmed_fee_msat = Some(skimmed_fee_msat); + } + + // Multiple HTLC allocations may describe the same channel pair. Collapse them into one record + // per pair while preserving the inferred totals above. + let mut grouped_allocations: Vec = Vec::new(); + for allocation in allocations { + let prev_channel_id = prev_htlcs[allocation.prev_htlc_index].channel_id; + let next_channel_id = next_htlcs[allocation.next_htlc_index].channel_id; + if let Some(existing) = grouped_allocations.iter_mut().find(|existing| { + prev_htlcs[existing.prev_htlc_index].channel_id == prev_channel_id + && next_htlcs[existing.next_htlc_index].channel_id == next_channel_id + }) { + existing.outbound_amount_forwarded_msat = existing + .outbound_amount_forwarded_msat + .saturating_add(allocation.outbound_amount_forwarded_msat); + existing.inbound_amount_forwarded_msat = existing + .inbound_amount_forwarded_msat + .saturating_add(allocation.inbound_amount_forwarded_msat); + existing.total_fee_earned_msat = + match (existing.total_fee_earned_msat, allocation.total_fee_earned_msat) { + (Some(existing), Some(additional)) => Some(existing.saturating_add(additional)), + _ => None, + }; + existing.skimmed_fee_msat = + match (existing.skimmed_fee_msat, allocation.skimmed_fee_msat) { + (Some(existing), Some(additional)) => Some(existing.saturating_add(additional)), + _ => None, + }; + existing.claim_from_onchain_tx |= allocation.claim_from_onchain_tx; + } else { + grouped_allocations.push(allocation); + } + } + + Some(grouped_allocations) +} + /// An event emitted by [`Node`], which should be handled by the user. /// /// [`Node`]: [`crate::Node`] @@ -353,11 +527,13 @@ impl_writeable_tlv_based_enum!(Event, (14, outbound_amount_forwarded_msat, option), (15, prev_htlcs, (default_value_vec, vec![HTLCLocator { channel_id: legacy_prev_channel_id.ok_or(lightning::ln::msgs::DecodeError::InvalidValue)?, + amount_msat: total_fee_earned_msat.and_then(|fee: u64| outbound_amount_forwarded_msat.map(|amount: u64| amount.saturating_add(fee))), user_channel_id: legacy_prev_user_channel_id.map(UserChannelId), node_id: legacy_prev_node_id, }])), (17, next_htlcs, (default_value_vec, vec![HTLCLocator { channel_id: legacy_next_channel_id.ok_or(lightning::ln::msgs::DecodeError::InvalidValue)?, + amount_msat: outbound_amount_forwarded_msat, user_channel_id: legacy_next_user_channel_id.map(UserChannelId), node_id: legacy_next_node_id, }])), @@ -535,6 +711,8 @@ where network_graph: Arc, liquidity_source: Arc>>, payment_store: Arc, + forwarded_payment_store: Arc, + channel_forwarding_stats_store: Arc, peer_store: Arc>, keys_manager: Arc, static_invoice_store: Option, @@ -556,6 +734,8 @@ where channel_manager: Arc, connection_manager: Arc>, output_sweeper: Arc, network_graph: Arc, liquidity_source: Arc>>, payment_store: Arc, + forwarded_payment_store: Arc, + channel_forwarding_stats_store: Arc, peer_store: Arc>, keys_manager: Arc, static_invoice_store: Option, onion_messenger: Arc, om_mailbox: Option>, prober: Option>, @@ -571,6 +751,8 @@ where network_graph, liquidity_source, payment_store, + forwarded_payment_store, + channel_forwarding_stats_store, peer_store, keys_manager, static_invoice_store, @@ -1490,6 +1672,178 @@ where .await; } + let all_htlc_amounts_known = prev_htlcs + .iter() + .chain(next_htlcs.iter()) + .all(|htlc| htlc.amount_msat.is_some()); + if !prev_htlcs.is_empty() && !next_htlcs.is_empty() && all_htlc_amounts_known { + let forwarded_at_timestamp = SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("current time should not be earlier than the Unix epoch") + .as_secs(); + + let mut inbound_stats_by_channel = HashMap::new(); + for (idx, prev_htlc) in prev_htlcs.iter().enumerate() { + let inbound_amount_msat = + prev_htlc.amount_msat.expect("all HTLC amounts were checked above"); + let inbound_stats = inbound_stats_by_channel + .entry(prev_htlc.channel_id) + .or_insert(ChannelForwardingStats { + channel_id: prev_htlc.channel_id, + counterparty_node_id: prev_htlc.node_id, + inbound_payments_forwarded: 1, + outbound_payments_forwarded: 0, + total_inbound_amount_msat: 0, + total_outbound_amount_msat: 0, + total_fee_earned_msat: total_fee_earned_msat.map(|_| 0), + total_skimmed_fee_msat: 0, + onchain_claims_count: 0, + first_forwarded_at_timestamp: forwarded_at_timestamp, + last_forwarded_at_timestamp: forwarded_at_timestamp, + }); + if inbound_stats.counterparty_node_id.is_none() { + inbound_stats.counterparty_node_id = prev_htlc.node_id; + } + inbound_stats.total_inbound_amount_msat += inbound_amount_msat; + if idx == prev_htlcs.len() - 1 { + if let (Some(total), Some(fee)) = ( + inbound_stats.total_fee_earned_msat.as_mut(), + total_fee_earned_msat, + ) { + *total += fee; + } + inbound_stats.total_skimmed_fee_msat += skimmed_fee_msat.unwrap_or(0); + } + } + + for inbound_stats in inbound_stats_by_channel.into_values() { + self.channel_forwarding_stats_store + .insert_or_update(inbound_stats) + .await + .map_err(|e| { + log_error!( + self.logger, + "Failed to update inbound channel forwarding stats: {e}" + ); + ReplayEvent() + })?; + } + + let mut outbound_stats_by_channel = HashMap::new(); + for (idx, next_htlc) in next_htlcs.iter().enumerate() { + let outbound_amount_msat = + next_htlc.amount_msat.expect("all HTLC amounts were checked above"); + let outbound_stats = outbound_stats_by_channel + .entry(next_htlc.channel_id) + .or_insert(ChannelForwardingStats { + channel_id: next_htlc.channel_id, + counterparty_node_id: next_htlc.node_id, + inbound_payments_forwarded: 0, + outbound_payments_forwarded: 1, + total_inbound_amount_msat: 0, + total_outbound_amount_msat: 0, + total_fee_earned_msat: Some(0), + total_skimmed_fee_msat: 0, + onchain_claims_count: 0, + first_forwarded_at_timestamp: forwarded_at_timestamp, + last_forwarded_at_timestamp: forwarded_at_timestamp, + }); + if outbound_stats.counterparty_node_id.is_none() { + outbound_stats.counterparty_node_id = next_htlc.node_id; + } + outbound_stats.total_outbound_amount_msat += outbound_amount_msat; + outbound_stats.onchain_claims_count += + if claim_from_onchain_tx && idx == 0 { 1 } else { 0 }; + } + + for outbound_stats in outbound_stats_by_channel.into_values() { + self.channel_forwarding_stats_store + .insert_or_update(outbound_stats) + .await + .map_err(|e| { + log_error!( + self.logger, + "Failed to update outbound channel forwarding stats: {e}" + ); + ReplayEvent() + })?; + } + + if matches!( + self.config.forwarded_payment_tracking_mode, + ForwardedPaymentTrackingMode::Detailed { .. } + ) { + if let Some(allocations) = forwarded_payment_allocations( + &prev_htlcs, + &next_htlcs, + total_fee_earned_msat, + skimmed_fee_msat, + claim_from_onchain_tx, + ) { + for allocation in allocations { + let prev_htlc = &prev_htlcs[allocation.prev_htlc_index]; + let next_htlc = &next_htlcs[allocation.next_htlc_index]; + self.forwarded_payment_store + .insert_with(|| { + let forwarded_at_timestamp = SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect( + "current time should not be earlier than the Unix epoch", + ) + .as_secs(); + ForwardedPaymentDetails { + id: ForwardedPaymentId( + self.keys_manager.get_secure_random_bytes(), + ), + prev_channel_id: prev_htlc.channel_id, + next_channel_id: next_htlc.channel_id, + prev_user_channel_id: prev_htlc + .user_channel_id + .map(UserChannelId), + next_user_channel_id: next_htlc + .user_channel_id + .map(UserChannelId), + prev_node_id: prev_htlc.node_id, + next_node_id: next_htlc.node_id, + inbound_amount_forwarded_msat: Some( + allocation.inbound_amount_forwarded_msat, + ), + total_fee_earned_msat: allocation.total_fee_earned_msat, + skimmed_fee_msat: allocation.skimmed_fee_msat, + claim_from_onchain_tx: allocation.claim_from_onchain_tx, + outbound_amount_forwarded_msat: Some( + allocation.outbound_amount_forwarded_msat, + ), + forwarded_at_timestamp, + } + }) + .await + .map_err(|e| { + log_error!( + self.logger, + "Failed to store forwarded payment: {e}" + ); + ReplayEvent() + })?; + } + } else { + log_debug!( + self.logger, + "Skipping detailed channel-pair forwarding stats for forward with {} inbound and {} outbound HTLCs because per-HTLC amounts are unavailable or invalid", + prev_htlcs.len(), + next_htlcs.len() + ); + } + } + } else if !prev_htlcs.is_empty() && !next_htlcs.is_empty() { + log_debug!( + self.logger, + "Skipping forwarding payment tracking for forward with {} inbound and {} outbound HTLCs because at least one per-HTLC amount is unavailable", + prev_htlcs.len(), + next_htlcs.len() + ); + } + let event = Event::PaymentForwarded { prev_htlcs: prev_htlcs.into_iter().map(HTLCLocator::from).collect(), next_htlcs: next_htlcs.into_iter().map(HTLCLocator::from).collect(), @@ -1972,6 +2326,469 @@ mod tests { use crate::payment::store::LSPS2Parameters; use crate::types::DynStoreWrapper; + fn ldk_htlc_locator(channel_byte: u8, amount_msat: Option) -> LdkHtlcLocator { + LdkHtlcLocator { + channel_id: ChannelId([channel_byte; 32]), + amount_msat, + user_channel_id: Some(channel_byte as u128), + node_id: None, + } + } + + #[test] + fn multi_htlc_forward_is_allocated_across_channel_pairs() { + let prev_htlcs = vec![ldk_htlc_locator(1, Some(600)), ldk_htlc_locator(2, Some(400))]; + let next_htlcs = vec![ldk_htlc_locator(3, Some(800))]; + + let allocations = + forwarded_payment_allocations(&prev_htlcs, &next_htlcs, Some(200), Some(40), true) + .unwrap(); + assert_eq!( + allocations, + vec![ + ForwardedPaymentAllocation { + prev_htlc_index: 0, + next_htlc_index: 0, + inbound_amount_forwarded_msat: 600, + total_fee_earned_msat: Some(0), + skimmed_fee_msat: Some(0), + claim_from_onchain_tx: true, + outbound_amount_forwarded_msat: 600, + }, + ForwardedPaymentAllocation { + prev_htlc_index: 1, + next_htlc_index: 0, + inbound_amount_forwarded_msat: 400, + total_fee_earned_msat: Some(200), + skimmed_fee_msat: Some(40), + claim_from_onchain_tx: true, + outbound_amount_forwarded_msat: 200, + }, + ], + ); + assert_eq!( + allocations + .iter() + .map(|allocation| allocation.outbound_amount_forwarded_msat) + .sum::(), + 800 + ); + assert_eq!( + allocations + .iter() + .map(|allocation| allocation.total_fee_earned_msat.unwrap()) + .sum::(), + 200 + ); + assert_eq!( + allocations.iter().map(|allocation| allocation.skimmed_fee_msat.unwrap()).sum::(), + 40 + ); + assert_eq!( + allocations.iter().filter(|allocation| allocation.claim_from_onchain_tx).count(), + 2 + ); + } + + #[test] + fn single_htlc_forward_preserves_all_fields() { + let prev_htlcs = vec![ldk_htlc_locator(1, Some(1_000))]; + let next_htlcs = vec![ldk_htlc_locator(2, Some(800))]; + assert_eq!( + forwarded_payment_allocations(&prev_htlcs, &next_htlcs, Some(200), Some(40), true), + Some(vec![ForwardedPaymentAllocation { + prev_htlc_index: 0, + next_htlc_index: 0, + inbound_amount_forwarded_msat: 1_000, + total_fee_earned_msat: Some(200), + skimmed_fee_msat: Some(40), + claim_from_onchain_tx: true, + outbound_amount_forwarded_msat: 800, + }]) + ); + } + + #[test] + fn multi_outbound_htlc_forward_is_allocated_across_channel_pairs() { + let prev_htlcs = vec![ldk_htlc_locator(1, Some(1_000))]; + let next_htlcs = vec![ldk_htlc_locator(2, Some(500)), ldk_htlc_locator(3, Some(300))]; + + let allocations = + forwarded_payment_allocations(&prev_htlcs, &next_htlcs, Some(200), None, false) + .unwrap(); + assert_eq!( + allocations, + vec![ + ForwardedPaymentAllocation { + prev_htlc_index: 0, + next_htlc_index: 0, + inbound_amount_forwarded_msat: 500, + total_fee_earned_msat: Some(0), + skimmed_fee_msat: None, + claim_from_onchain_tx: false, + outbound_amount_forwarded_msat: 500, + }, + ForwardedPaymentAllocation { + prev_htlc_index: 0, + next_htlc_index: 1, + inbound_amount_forwarded_msat: 500, + total_fee_earned_msat: Some(200), + skimmed_fee_msat: None, + claim_from_onchain_tx: false, + outbound_amount_forwarded_msat: 300, + }, + ], + ); + } + + #[test] + fn multi_htlc_forward_preserves_inbound_amounts_with_unknown_fee() { + let prev_htlcs = vec![ldk_htlc_locator(1, Some(600)), ldk_htlc_locator(2, Some(400))]; + let next_htlcs = vec![ldk_htlc_locator(3, Some(800))]; + + let allocations = + forwarded_payment_allocations(&prev_htlcs, &next_htlcs, None, None, false).unwrap(); + assert_eq!( + allocations, + vec![ + ForwardedPaymentAllocation { + prev_htlc_index: 0, + next_htlc_index: 0, + inbound_amount_forwarded_msat: 600, + total_fee_earned_msat: None, + skimmed_fee_msat: None, + claim_from_onchain_tx: false, + outbound_amount_forwarded_msat: 600, + }, + ForwardedPaymentAllocation { + prev_htlc_index: 1, + next_htlc_index: 0, + inbound_amount_forwarded_msat: 400, + total_fee_earned_msat: None, + skimmed_fee_msat: None, + claim_from_onchain_tx: false, + outbound_amount_forwarded_msat: 200, + }, + ], + ); + } + + #[test] + fn multi_htlc_forward_groups_repeated_channel_pairs() { + let prev_htlcs = vec![ldk_htlc_locator(1, Some(600)), ldk_htlc_locator(1, Some(400))]; + let next_htlcs = vec![ldk_htlc_locator(2, Some(800))]; + + let allocations = + forwarded_payment_allocations(&prev_htlcs, &next_htlcs, Some(200), None, false) + .unwrap(); + assert_eq!( + allocations, + vec![ForwardedPaymentAllocation { + prev_htlc_index: 0, + next_htlc_index: 0, + inbound_amount_forwarded_msat: 1_000, + total_fee_earned_msat: Some(200), + skimmed_fee_msat: None, + claim_from_onchain_tx: false, + outbound_amount_forwarded_msat: 800, + }], + ); + } + + #[test] + fn multi_htlc_forward_requires_all_locator_amounts() { + let prev_htlcs = vec![ldk_htlc_locator(1, Some(600)), ldk_htlc_locator(2, None)]; + let next_htlcs = vec![ldk_htlc_locator(3, Some(800))]; + assert_eq!( + forwarded_payment_allocations(&prev_htlcs, &next_htlcs, Some(200), None, false), + None + ); + } + + #[test] + fn multi_inbound_and_outbound_htlcs_are_allocated_across_channel_pairs() { + let prev_htlcs = vec![ldk_htlc_locator(1, Some(600)), ldk_htlc_locator(2, Some(400))]; + let next_htlcs = vec![ldk_htlc_locator(3, Some(500)), ldk_htlc_locator(4, Some(300))]; + assert_eq!( + forwarded_payment_allocations(&prev_htlcs, &next_htlcs, Some(200), None, false), + Some(vec![ + ForwardedPaymentAllocation { + prev_htlc_index: 0, + next_htlc_index: 0, + inbound_amount_forwarded_msat: 500, + total_fee_earned_msat: Some(0), + skimmed_fee_msat: None, + claim_from_onchain_tx: false, + outbound_amount_forwarded_msat: 500, + }, + ForwardedPaymentAllocation { + prev_htlc_index: 0, + next_htlc_index: 1, + inbound_amount_forwarded_msat: 100, + total_fee_earned_msat: Some(0), + skimmed_fee_msat: None, + claim_from_onchain_tx: false, + outbound_amount_forwarded_msat: 100, + }, + ForwardedPaymentAllocation { + prev_htlc_index: 1, + next_htlc_index: 1, + inbound_amount_forwarded_msat: 400, + total_fee_earned_msat: Some(200), + skimmed_fee_msat: None, + claim_from_onchain_tx: false, + outbound_amount_forwarded_msat: 200, + }, + ]) + ); + } + + #[test] + fn equal_htlc_counts_are_split_by_amount() { + let prev_htlcs = vec![ + ldk_htlc_locator(1, Some(200)), + ldk_htlc_locator(2, Some(300)), + ldk_htlc_locator(3, Some(500)), + ]; + let next_htlcs = vec![ + ldk_htlc_locator(4, Some(500)), + ldk_htlc_locator(5, Some(150)), + ldk_htlc_locator(6, Some(350)), + ]; + assert_eq!( + forwarded_payment_allocations(&prev_htlcs, &next_htlcs, Some(0), None, false), + Some(vec![ + ForwardedPaymentAllocation { + prev_htlc_index: 0, + next_htlc_index: 0, + inbound_amount_forwarded_msat: 200, + total_fee_earned_msat: Some(0), + skimmed_fee_msat: None, + claim_from_onchain_tx: false, + outbound_amount_forwarded_msat: 200, + }, + ForwardedPaymentAllocation { + prev_htlc_index: 1, + next_htlc_index: 0, + inbound_amount_forwarded_msat: 300, + total_fee_earned_msat: Some(0), + skimmed_fee_msat: None, + claim_from_onchain_tx: false, + outbound_amount_forwarded_msat: 300, + }, + ForwardedPaymentAllocation { + prev_htlc_index: 2, + next_htlc_index: 1, + inbound_amount_forwarded_msat: 150, + total_fee_earned_msat: Some(0), + skimmed_fee_msat: None, + claim_from_onchain_tx: false, + outbound_amount_forwarded_msat: 150, + }, + ForwardedPaymentAllocation { + prev_htlc_index: 2, + next_htlc_index: 2, + inbound_amount_forwarded_msat: 350, + total_fee_earned_msat: Some(0), + skimmed_fee_msat: None, + claim_from_onchain_tx: false, + outbound_amount_forwarded_msat: 350, + }, + ]) + ); + } + + #[test] + fn unmatched_inbound_amount_spans_multiple_htlcs() { + let prev_htlcs = vec![ + ldk_htlc_locator(1, Some(300)), + ldk_htlc_locator(2, Some(400)), + ldk_htlc_locator(3, Some(500)), + ]; + let next_htlcs = vec![ldk_htlc_locator(4, Some(500))]; + assert_eq!( + forwarded_payment_allocations(&prev_htlcs, &next_htlcs, Some(700), None, false), + Some(vec![ + ForwardedPaymentAllocation { + prev_htlc_index: 0, + next_htlc_index: 0, + inbound_amount_forwarded_msat: 300, + total_fee_earned_msat: Some(0), + skimmed_fee_msat: None, + claim_from_onchain_tx: false, + outbound_amount_forwarded_msat: 300, + }, + ForwardedPaymentAllocation { + prev_htlc_index: 1, + next_htlc_index: 0, + inbound_amount_forwarded_msat: 400, + total_fee_earned_msat: Some(0), + skimmed_fee_msat: None, + claim_from_onchain_tx: false, + outbound_amount_forwarded_msat: 200, + }, + ForwardedPaymentAllocation { + prev_htlc_index: 2, + next_htlc_index: 0, + inbound_amount_forwarded_msat: 500, + total_fee_earned_msat: Some(700), + skimmed_fee_msat: None, + claim_from_onchain_tx: false, + outbound_amount_forwarded_msat: 0, + }, + ]) + ); + } + + #[test] + fn unmatched_outbound_amount_spans_multiple_htlcs() { + let prev_htlcs = vec![ldk_htlc_locator(1, Some(500))]; + let next_htlcs = vec![ + ldk_htlc_locator(2, Some(200)), + ldk_htlc_locator(3, Some(400)), + ldk_htlc_locator(4, Some(300)), + ]; + assert_eq!( + forwarded_payment_allocations(&prev_htlcs, &next_htlcs, None, None, false), + Some(vec![ + ForwardedPaymentAllocation { + prev_htlc_index: 0, + next_htlc_index: 0, + inbound_amount_forwarded_msat: 200, + total_fee_earned_msat: None, + skimmed_fee_msat: None, + claim_from_onchain_tx: false, + outbound_amount_forwarded_msat: 200, + }, + ForwardedPaymentAllocation { + prev_htlc_index: 0, + next_htlc_index: 1, + inbound_amount_forwarded_msat: 300, + total_fee_earned_msat: None, + skimmed_fee_msat: None, + claim_from_onchain_tx: false, + outbound_amount_forwarded_msat: 400, + }, + ForwardedPaymentAllocation { + prev_htlc_index: 0, + next_htlc_index: 2, + inbound_amount_forwarded_msat: 0, + total_fee_earned_msat: None, + skimmed_fee_msat: None, + claim_from_onchain_tx: false, + outbound_amount_forwarded_msat: 300, + }, + ]) + ); + } + + #[test] + fn non_adjacent_repeated_channel_pairs_are_grouped() { + let prev_htlcs = vec![ + ldk_htlc_locator(1, Some(200)), + ldk_htlc_locator(2, Some(300)), + ldk_htlc_locator(1, Some(500)), + ]; + let next_htlcs = vec![ + ldk_htlc_locator(3, Some(500)), + ldk_htlc_locator(4, Some(150)), + ldk_htlc_locator(3, Some(350)), + ]; + assert_eq!( + forwarded_payment_allocations(&prev_htlcs, &next_htlcs, Some(0), None, false), + Some(vec![ + ForwardedPaymentAllocation { + prev_htlc_index: 0, + next_htlc_index: 0, + inbound_amount_forwarded_msat: 550, + total_fee_earned_msat: Some(0), + skimmed_fee_msat: None, + claim_from_onchain_tx: false, + outbound_amount_forwarded_msat: 550, + }, + ForwardedPaymentAllocation { + prev_htlc_index: 1, + next_htlc_index: 0, + inbound_amount_forwarded_msat: 300, + total_fee_earned_msat: Some(0), + skimmed_fee_msat: None, + claim_from_onchain_tx: false, + outbound_amount_forwarded_msat: 300, + }, + ForwardedPaymentAllocation { + prev_htlc_index: 2, + next_htlc_index: 1, + inbound_amount_forwarded_msat: 150, + total_fee_earned_msat: Some(0), + skimmed_fee_msat: None, + claim_from_onchain_tx: false, + outbound_amount_forwarded_msat: 150, + }, + ]) + ); + } + + #[test] + fn unequal_htlc_counts_are_allocated_fifo() { + let prev_htlcs = vec![ldk_htlc_locator(1, Some(300)), ldk_htlc_locator(2, Some(600))]; + let next_htlcs = vec![ + ldk_htlc_locator(3, Some(200)), + ldk_htlc_locator(4, Some(300)), + ldk_htlc_locator(5, Some(200)), + ]; + assert_eq!( + forwarded_payment_allocations(&prev_htlcs, &next_htlcs, Some(200), None, false), + Some(vec![ + ForwardedPaymentAllocation { + prev_htlc_index: 0, + next_htlc_index: 0, + inbound_amount_forwarded_msat: 200, + total_fee_earned_msat: Some(0), + skimmed_fee_msat: None, + claim_from_onchain_tx: false, + outbound_amount_forwarded_msat: 200, + }, + ForwardedPaymentAllocation { + prev_htlc_index: 0, + next_htlc_index: 1, + inbound_amount_forwarded_msat: 100, + total_fee_earned_msat: Some(0), + skimmed_fee_msat: None, + claim_from_onchain_tx: false, + outbound_amount_forwarded_msat: 100, + }, + ForwardedPaymentAllocation { + prev_htlc_index: 1, + next_htlc_index: 1, + inbound_amount_forwarded_msat: 200, + total_fee_earned_msat: Some(0), + skimmed_fee_msat: None, + claim_from_onchain_tx: false, + outbound_amount_forwarded_msat: 200, + }, + ForwardedPaymentAllocation { + prev_htlc_index: 1, + next_htlc_index: 2, + inbound_amount_forwarded_msat: 400, + total_fee_earned_msat: Some(200), + skimmed_fee_msat: None, + claim_from_onchain_tx: false, + outbound_amount_forwarded_msat: 200, + }, + ]) + ); + } + + #[test] + fn single_htlc_forward_with_unknown_amount_is_skipped() { + let prev_htlcs = vec![ldk_htlc_locator(1, None)]; + let next_htlcs = vec![ldk_htlc_locator(2, None)]; + assert_eq!( + forwarded_payment_allocations(&prev_htlcs, &next_htlcs, Some(20), None, false), + None + ); + } + #[test] fn lsps2_payment_metadata_decodes_total_fee_limit() { let metadata = PaymentMetadata { @@ -2067,6 +2884,14 @@ mod tests { counterparty_node_id: Option, reason: Option, }, + PaymentForwarded { + prev_htlcs: Vec, + next_htlcs: Vec, + total_fee_earned_msat: Option, + skimmed_fee_msat: Option, + claim_from_onchain_tx: bool, + outbound_amount_forwarded_msat: Option, + }, } impl_writeable_tlv_based_enum!(LegacyEvent, @@ -2076,6 +2901,14 @@ mod tests { (2, user_channel_id, required), (3, reason, upgradable_option), }, + (7, PaymentForwarded) => { + (8, total_fee_earned_msat, option), + (10, skimmed_fee_msat, option), + (12, claim_from_onchain_tx, required), + (14, outbound_amount_forwarded_msat, option), + (15, prev_htlcs, (default_value_vec, Vec::new())), + (17, next_htlcs, (default_value_vec, Vec::new())), + }, ); fn encode_legacy_event_queue(event: LegacyEvent) -> Vec { @@ -2137,6 +2970,54 @@ mod tests { assert!(res.is_err()); } + #[test] + fn event_queue_reads_legacy_multi_htlc_forward() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let logger = Arc::new(TestLogger::new()); + let prev_htlcs = vec![ + HTLCLocator::from(ldk_htlc_locator(1, Some(600))), + HTLCLocator::from(ldk_htlc_locator(2, Some(400))), + ]; + let next_htlcs = vec![HTLCLocator::from(ldk_htlc_locator(3, Some(800)))]; + let legacy_event = LegacyEvent::PaymentForwarded { + prev_htlcs: prev_htlcs.clone(), + next_htlcs: next_htlcs.clone(), + total_fee_earned_msat: Some(200), + skimmed_fee_msat: None, + claim_from_onchain_tx: false, + outbound_amount_forwarded_msat: Some(800), + }; + let persisted_bytes = encode_legacy_event_queue(legacy_event); + + let event_queue = + EventQueue::read(&mut &persisted_bytes[..], (Arc::clone(&store), logger)).unwrap(); + assert_eq!( + event_queue.next_event(), + Some(Event::PaymentForwarded { + prev_htlcs, + next_htlcs, + total_fee_earned_msat: Some(200), + skimmed_fee_msat: None, + claim_from_onchain_tx: false, + outbound_amount_forwarded_msat: Some(800), + }) + ); + } + + #[test] + fn payment_forwarded_event_roundtrips() { + let event = Event::PaymentForwarded { + prev_htlcs: vec![HTLCLocator::from(ldk_htlc_locator(1, Some(1_000)))], + next_htlcs: vec![HTLCLocator::from(ldk_htlc_locator(2, Some(800)))], + total_fee_earned_msat: Some(200), + skimmed_fee_msat: None, + claim_from_onchain_tx: false, + outbound_amount_forwarded_msat: Some(800), + }; + + assert_eq!(Event::read(&mut &event.encode()[..]).unwrap(), event); + } + #[tokio::test] async fn event_queue_concurrency() { let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); diff --git a/src/ffi/types.rs b/src/ffi/types.rs index c6b48dc961..8adba6241a 100644 --- a/src/ffi/types.rs +++ b/src/ffi/types.rs @@ -149,8 +149,9 @@ pub use crate::entropy::{generate_entropy_mnemonic, NodeEntropy, WordCount}; use crate::error::Error; pub use crate::liquidity::LSPS1OrderStatus; pub use crate::logger::{LogLevel, LogRecord, LogWriter}; +use crate::payment::store::{ChannelPairStatsId, ForwardedPaymentId}; pub use crate::probing::ProbingConfig; -use crate::{hex_utils, SocketAddress, UserChannelId}; +use crate::{hex_utils, PageToken, SocketAddress, UserChannelId}; uniffi::custom_type!(PublicKey, String, { remote, @@ -910,6 +911,22 @@ uniffi::custom_type!(PaymentId, String, { }, }); +uniffi::custom_type!(ForwardedPaymentId, String, { + remote, + try_lift: |val| { + if let Some(bytes_vec) = hex_utils::to_vec(&val) { + let bytes_res = bytes_vec.try_into(); + if let Ok(bytes) = bytes_res { + return Ok(ForwardedPaymentId(bytes)); + } + } + Err(Error::InvalidPaymentId.into()) + }, + lower: |obj| { + hex_utils::to_string(&obj.0) + }, +}); + uniffi::custom_type!(PaymentHash, String, { remote, try_lift: |val| { @@ -975,6 +992,33 @@ uniffi::custom_type!(ChannelId, String, { } }); +uniffi::custom_type!(ChannelPairStatsId, String, { + remote, + try_lift: |val| { + if let Some(hex_vec) = hex_utils::to_vec(&val) { + if hex_vec.len() == 80 { + let mut id = [0u8; 80]; + id.copy_from_slice(&hex_vec[..]); + return Ok(ChannelPairStatsId::from_bytes(id)); + } + } + Err(Error::InvalidChannelId.into()) + }, + lower: |obj| { + hex_utils::to_string(obj.as_bytes()) + } +}); + +uniffi::custom_type!(PageToken, String, { + remote, + try_lift: |val| { + Ok(PageToken::new(val)) + }, + lower: |obj| { + obj.to_string() + } +}); + uniffi::custom_type!(UserChannelId, String, { remote, try_lift: |val| { diff --git a/src/io/mod.rs b/src/io/mod.rs index a01aa59a83..6cad197138 100644 --- a/src/io/mod.rs +++ b/src/io/mod.rs @@ -33,6 +33,25 @@ pub(crate) const PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE: &str = ""; pub(crate) const PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE: &str = "pending_payments"; pub(crate) const PENDING_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE: &str = ""; +/// The forwarded payment information will be persisted under this prefix. +pub(crate) const FORWARDED_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE: &str = "forwarded_payments"; +pub(crate) const FORWARDED_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE: &str = ""; + +/// The channel forwarding stats will be persisted under this prefix. +pub(crate) const CHANNEL_FORWARDING_STATS_PERSISTENCE_PRIMARY_NAMESPACE: &str = + "channel_forwarding_stats"; +pub(crate) const CHANNEL_FORWARDING_STATS_PERSISTENCE_SECONDARY_NAMESPACE: &str = ""; + +/// The channel pair forwarding stats will be persisted under this prefix. +pub(crate) const CHANNEL_PAIR_FORWARDING_STATS_PERSISTENCE_PRIMARY_NAMESPACE: &str = + "channel_pair_forwarding_stats"; +pub(crate) const CHANNEL_PAIR_FORWARDING_STATS_PERSISTENCE_SECONDARY_NAMESPACE: &str = ""; + +/// The immutable forwarded-payment bucket size will be persisted under this key. +pub(crate) const FORWARDED_PAYMENT_BUCKET_SIZE_PRIMARY_NAMESPACE: &str = ""; +pub(crate) const FORWARDED_PAYMENT_BUCKET_SIZE_SECONDARY_NAMESPACE: &str = ""; +pub(crate) const FORWARDED_PAYMENT_BUCKET_SIZE_KEY: &str = "forwarded_payment_bucket_size"; + /// The node metrics will be persisted under this key. pub(crate) const NODE_METRICS_PRIMARY_NAMESPACE: &str = ""; pub(crate) const NODE_METRICS_SECONDARY_NAMESPACE: &str = ""; diff --git a/src/io/utils.rs b/src/io/utils.rs index 4657688f51..1df84f3398 100644 --- a/src/io/utils.rs +++ b/src/io/utils.rs @@ -43,7 +43,9 @@ use crate::chain::ChainSource; use crate::config::WALLET_KEYS_SEED_LEN; use crate::fee_estimator::OnchainFeeEstimator; use crate::io::{ - NODE_METRICS_KEY, NODE_METRICS_PRIMARY_NAMESPACE, NODE_METRICS_SECONDARY_NAMESPACE, + FORWARDED_PAYMENT_BUCKET_SIZE_KEY, FORWARDED_PAYMENT_BUCKET_SIZE_PRIMARY_NAMESPACE, + FORWARDED_PAYMENT_BUCKET_SIZE_SECONDARY_NAMESPACE, NODE_METRICS_KEY, + NODE_METRICS_PRIMARY_NAMESPACE, NODE_METRICS_SECONDARY_NAMESPACE, }; use crate::logger::{log_error, LdkLogger, Logger}; use crate::peer_store::PeerStore; @@ -335,6 +337,50 @@ where }) } +pub(crate) async fn read_forwarded_payment_bucket_size( + kv_store: &DynStore, logger: L, +) -> Result +where + L::Target: LdkLogger, +{ + let reader = KVStore::read( + &*kv_store, + FORWARDED_PAYMENT_BUCKET_SIZE_PRIMARY_NAMESPACE, + FORWARDED_PAYMENT_BUCKET_SIZE_SECONDARY_NAMESPACE, + FORWARDED_PAYMENT_BUCKET_SIZE_KEY, + ) + .await?; + let bucket_size_secs = u64::read(&mut &*reader).map_err(|e| { + log_error!(logger, "Failed to deserialize forwarded-payment bucket size: {}", e); + std::io::Error::new( + std::io::ErrorKind::InvalidData, + "Failed to deserialize forwarded-payment bucket size", + ) + })?; + if bucket_size_secs == 0 { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + "Forwarded-payment bucket size must be non-zero", + )); + } + Ok(bucket_size_secs) +} + +pub(crate) async fn write_forwarded_payment_bucket_size( + kv_store: &DynStore, bucket_size_secs: u64, +) -> Result<(), std::io::Error> { + debug_assert!(bucket_size_secs > 0); + KVStore::write( + &*kv_store, + FORWARDED_PAYMENT_BUCKET_SIZE_PRIMARY_NAMESPACE, + FORWARDED_PAYMENT_BUCKET_SIZE_SECONDARY_NAMESPACE, + FORWARDED_PAYMENT_BUCKET_SIZE_KEY, + bucket_size_secs.encode(), + ) + .await?; + Ok(()) +} + /// Take a write lock on `node_metrics`, apply `update`, and persist the result to `kv_store`. pub(crate) async fn update_and_persist_node_metrics( node_metrics: &PersistedNodeMetrics, kv_store: &DynStore, logger: L, @@ -718,19 +764,36 @@ fn recover_incomplete_fs_store_migration(storage_dir_path: &Path) -> Result<(), mod tests { use std::fs; use std::path::{Path, PathBuf}; + use std::sync::Arc; use lightning::util::persist::{migrate_kv_store_data_async, KVStore}; use lightning_persister::fs_store::v1::FilesystemStore; use lightning_persister::fs_store::v2::FilesystemStoreV2; + use crate::io::test_utils::InMemoryStore; + use crate::logger::Logger; + use crate::types::{DynStore, DynStoreWrapper}; + use super::test_utils::random_storage_path; - use super::{open_or_migrate_fs_store, read_or_generate_seed_file}; + use super::{ + open_or_migrate_fs_store, read_forwarded_payment_bucket_size, read_or_generate_seed_file, + write_forwarded_payment_bucket_size, + }; const TEST_PRIMARY_NAMESPACE: &str = "test_primary_namespace"; const TEST_SECONDARY_NAMESPACE: &str = "test_secondary_namespace"; const TEST_KEY: &str = "test_key"; const TEST_VALUE: &[u8] = b"test_value"; + #[tokio::test] + async fn forwarded_payment_bucket_size_roundtrips() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let logger = Arc::new(Logger::new_log_facade()); + + write_forwarded_payment_bucket_size(&*store, 900).await.unwrap(); + assert_eq!(read_forwarded_payment_bucket_size(&*store, logger).await.unwrap(), 900); + } + #[test] fn generated_seed_is_readable() { let mut rand_path = random_storage_path(); diff --git a/src/lib.rs b/src/lib.rs index acfcbc0d48..6f6cd4aa10 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -158,6 +158,7 @@ use lightning::ln::peer_handler::CustomMessageHandler; use lightning::routing::gossip::NodeAlias; use lightning::sign::EntropySource; use lightning::util::persist::KVStore; +pub use lightning::util::persist::PageToken; use lightning::util::wallet_utils::{Input, Wallet as LdkWallet}; use lightning_background_processor::process_events_async; pub use lightning_invoice; @@ -169,9 +170,13 @@ use lnurl_auth::LnurlAuth; use logger::{log_debug, log_error, log_info, log_trace, LdkLogger, Logger}; use payment::asynchronous::om_mailbox::OnionMessageMailbox; use payment::asynchronous::static_invoice_store::StaticInvoiceStore; +use payment::store::{ + aggregate_channel_pair_stats as aggregate_channel_pair_stats_impl, + aggregate_expired_forwarded_payments, run_forwarded_payment_aggregation, +}; use payment::{ - Bolt11Payment, Bolt12Payment, OnchainPayment, PaymentDetails, SpontaneousPayment, - UnifiedPayment, + Bolt11Payment, Bolt12Payment, ChannelPairForwardingStats, Forwarding, OnchainPayment, + PaymentDetails, SpontaneousPayment, UnifiedPayment, }; use peer_store::{PeerInfo, PeerStore}; #[cfg(feature = "uniffi")] @@ -180,7 +185,8 @@ use probing::{run_prober, Prober}; use runtime::Runtime; pub use tokio; use types::{ - Broadcaster, BumpTransactionEventHandler, ChainMonitor, ChannelManager, DynStore, Graph, + Broadcaster, BumpTransactionEventHandler, ChainMonitor, ChannelForwardingStatsStore, + ChannelManager, ChannelPairForwardingStatsStore, DynStore, ForwardedPaymentStore, Graph, HRNResolver, KeysManager, OnionMessenger, PaymentStore, PeerManager, Router, Scorer, Sweeper, Wallet, }; @@ -249,6 +255,10 @@ pub struct Node { scorer: Arc>, peer_store: Arc>>, payment_store: Arc, + forwarded_payment_store: Arc, + channel_forwarding_stats_store: Arc, + channel_pair_forwarding_stats_store: Arc, + forwarded_payment_bucket_size_secs: Option, lnurl_auth: Arc, is_running: Arc>, node_metrics: Arc, @@ -295,6 +305,39 @@ impl Node { let chain_source = Arc::clone(&self.chain_source); self.runtime.block_on(async move { chain_source.update_fee_rate_estimates().await })?; + if let Some(forwarded_payment_bucket_size_secs) = self.forwarded_payment_bucket_size_secs { + let forwarded_payment_store = Arc::clone(&self.forwarded_payment_store); + let channel_pair_stats_store = Arc::clone(&self.channel_pair_forwarding_stats_store); + let logger = Arc::clone(&self.logger); + self.runtime.block_on(async move { + match aggregate_expired_forwarded_payments( + &forwarded_payment_store, + &channel_pair_stats_store, + forwarded_payment_bucket_size_secs, + &logger, + ) + .await + { + Ok((pair_count, payment_count)) if pair_count > 0 => { + log_info!( + logger, + "Aggregated {payment_count} forwarded payments into {pair_count} channel pair buckets" + ); + }, + Ok((0, payment_count)) if payment_count > 0 => { + log_info!( + logger, + "Removed {payment_count} forwarded payment details from previously aggregated buckets" + ); + }, + Err(e) => { + log_error!(logger, "Startup forwarded payment aggregation failed: {e}") + }, + _ => {}, + } + }); + } + // Spawn background task continuously syncing onchain, lightning, and fee rate cache. let stop_sync_receiver = self.stop_sender.subscribe(); let chain_source = Arc::clone(&self.chain_source); @@ -587,6 +630,23 @@ impl Node { chain_source.continuously_process_broadcast_queue(stop_tx_bcast).await }); + if let Some(forwarded_payment_bucket_size_secs) = self.forwarded_payment_bucket_size_secs { + let stop_aggregation = self.stop_sender.subscribe(); + let forwarded_payment_store = Arc::clone(&self.forwarded_payment_store); + let channel_pair_stats_store = Arc::clone(&self.channel_pair_forwarding_stats_store); + let logger = Arc::clone(&self.logger); + self.runtime.spawn_cancellable_background_task(async move { + run_forwarded_payment_aggregation( + stop_aggregation, + forwarded_payment_store, + channel_pair_stats_store, + forwarded_payment_bucket_size_secs, + logger, + ) + .await; + }); + } + let bump_tx_event_handler = Arc::new(BumpTransactionEventHandler::new( Arc::clone(&self.tx_broadcaster), Arc::new(LdkWallet::new(Arc::clone(&self.wallet), Arc::clone(&self.logger))), @@ -611,6 +671,8 @@ impl Node { Arc::clone(&self.network_graph), Arc::clone(&self.liquidity_source), Arc::clone(&self.payment_store), + Arc::clone(&self.forwarded_payment_store), + Arc::clone(&self.channel_forwarding_stats_store), Arc::clone(&self.peer_store), Arc::clone(&self.keys_manager), static_invoice_store, @@ -1115,6 +1177,30 @@ impl Node { )) } + /// Returns a handler allowing to query forwarded payments and forwarding statistics. + #[cfg(not(feature = "uniffi"))] + pub fn forwarding(&self) -> Forwarding { + Forwarding::new( + Arc::clone(&self.runtime), + Arc::clone(&self.forwarded_payment_store), + Arc::clone(&self.channel_forwarding_stats_store), + Arc::clone(&self.channel_pair_forwarding_stats_store), + Arc::clone(&self.config), + ) + } + + /// Returns a handler allowing to query forwarded payments and forwarding statistics. + #[cfg(feature = "uniffi")] + pub fn forwarding(&self) -> Arc { + Arc::new(Forwarding::new( + Arc::clone(&self.runtime), + Arc::clone(&self.forwarded_payment_store), + Arc::clone(&self.channel_forwarding_stats_store), + Arc::clone(&self.channel_pair_forwarding_stats_store), + Arc::clone(&self.config), + )) + } + /// Authenticates the user via [LNURL-auth] for the given LNURL string. /// /// [LNURL-auth]: https://github.com/lnurl/luds/blob/luds/04.md @@ -2435,6 +2521,18 @@ pub(crate) fn new_channel_anchor_reserve_sats( }) } +/// Aggregates multiple channel-pair statistics buckets into cumulative totals. +/// +/// The returned bucket spans from the earliest input bucket start through the latest input bucket +/// end, including any gaps between buckets. +/// +/// Returns `None` if `buckets` is empty or contains statistics for different channel pairs. +pub fn aggregate_channel_pair_stats( + buckets: &[ChannelPairForwardingStats], +) -> Option { + aggregate_channel_pair_stats_impl(buckets) +} + #[cfg(test)] mod tests { use lightning::util::ser::{Readable, Writeable}; diff --git a/src/payment/forwarding.rs b/src/payment/forwarding.rs new file mode 100644 index 0000000000..97bca874d7 --- /dev/null +++ b/src/payment/forwarding.rs @@ -0,0 +1,200 @@ +// This file is Copyright its original authors, visible in version control history. +// +// This file is licensed under the Apache License, Version 2.0 or the MIT license , at your option. You may not use this file except in +// accordance with one or both of these licenses. + +//! Holds a handler allowing to query forwarded payments and forwarding statistics. + +use std::sync::Arc; + +use lightning::ln::types::ChannelId; +use lightning::util::persist::PageToken; + +use crate::config::{Config, ForwardedPaymentTrackingMode}; +use crate::error::Error; +use crate::payment::store::{ + ChannelForwardingStats, ChannelPairForwardingStats, ForwardedPaymentDetails, ForwardedPaymentId, +}; +use crate::runtime::Runtime; +use crate::types::{ + ChannelForwardingStatsStore, ChannelPairForwardingStatsStore, ForwardedPaymentStore, +}; + +/// A page of forwarded payments returned from a paginated listing. +#[derive(Clone, Debug, PartialEq, Eq)] +#[cfg_attr(feature = "uniffi", derive(uniffi::Record))] +pub struct ForwardedPaymentDetailsPage { + /// Forwarded payments in this page. + pub payments: Vec, + /// Token to pass to the next call to continue listing, if another page exists. + pub next_page_token: Option, +} + +/// A page of channel forwarding statistics returned from a paginated listing. +#[derive(Clone, Debug, PartialEq, Eq)] +#[cfg_attr(feature = "uniffi", derive(uniffi::Record))] +pub struct ChannelForwardingStatsPage { + /// Channel forwarding statistics in this page. + pub stats: Vec, + /// Token to pass to the next call to continue listing, if another page exists. + pub next_page_token: Option, +} + +/// A page of channel-pair forwarding statistics returned from a paginated listing. +#[derive(Clone, Debug, PartialEq, Eq)] +#[cfg_attr(feature = "uniffi", derive(uniffi::Record))] +pub struct ChannelPairForwardingStatsPage { + /// Channel-pair forwarding statistics in this page. + pub stats: Vec, + /// Token to pass to the next call to continue listing, if another page exists. + pub next_page_token: Option, +} + +/// A handler allowing to query forwarded payments and forwarding statistics. +/// +/// Should be retrieved by calling [`Node::forwarding`]. +/// +/// [`Node::forwarding`]: crate::Node::forwarding +#[cfg_attr(feature = "uniffi", derive(uniffi::Object))] +pub struct Forwarding { + runtime: Arc, + forwarded_payment_store: Arc, + channel_forwarding_stats_store: Arc, + channel_pair_forwarding_stats_store: Arc, + config: Arc, +} + +impl Forwarding { + pub(crate) fn new( + runtime: Arc, forwarded_payment_store: Arc, + channel_forwarding_stats_store: Arc, + channel_pair_forwarding_stats_store: Arc, + config: Arc, + ) -> Self { + Self { + runtime, + forwarded_payment_store, + channel_forwarding_stats_store, + channel_pair_forwarding_stats_store, + config, + } + } + + /// Retrieves all forwarded payments that match the given predicate. + pub fn list_payments_with_filter bool>( + &self, f: F, + ) -> Vec { + self.forwarded_payment_store.list_filter(f) + } + + /// Retrieves all channel forwarding statistics that match the given predicate. + pub fn list_channel_stats_with_filter bool>( + &self, f: F, + ) -> Vec { + self.channel_forwarding_stats_store.list_filter(f) + } + + /// Retrieves all channel pair forwarding statistics that match the given predicate. + pub fn list_channel_pair_stats_with_filter bool>( + &self, f: F, + ) -> Vec { + self.channel_pair_forwarding_stats_store.list_filter(f) + } + + /// Retrieves pages of channel-pair forwarding statistics until the given filter produces a + /// non-empty result page or the underlying store is exhausted. + fn list_channel_pair_stats_filtered_page bool>( + &self, mut page_token: Option, mut f: F, + ) -> Result { + loop { + let (mut stats, next_page_token) = self + .runtime + .block_on(self.channel_pair_forwarding_stats_store.list_page(page_token))?; + stats.retain(|stats| f(stats)); + if !stats.is_empty() || next_page_token.is_none() { + stats.sort_by_key(|stats| stats.bucket_start_timestamp); + return Ok(ChannelPairForwardingStatsPage { stats, next_page_token }); + } + page_token = next_page_token; + } + } +} + +#[cfg_attr(feature = "uniffi", uniffi::export)] +impl Forwarding { + /// Returns the configured forwarded payment tracking mode. + pub fn tracking_mode(&self) -> ForwardedPaymentTrackingMode { + self.config.forwarded_payment_tracking_mode + } + + /// Retrieve the details of a specific forwarded payment with the given id. + pub fn payment( + &self, forwarded_payment_id: &ForwardedPaymentId, + ) -> Option { + self.forwarded_payment_store.get(forwarded_payment_id) + } + + /// Retrieves a page of forwarded payments from the underlying paginated store. + pub fn list_payments( + &self, page_token: Option, + ) -> Result { + let (payments, next_page_token) = + self.runtime.block_on(self.forwarded_payment_store.list_page(page_token))?; + Ok(ForwardedPaymentDetailsPage { payments, next_page_token }) + } + + /// Retrieve the forwarding statistics for a specific channel. + pub fn channel_stats(&self, channel_id: &ChannelId) -> Option { + self.channel_forwarding_stats_store.get(channel_id) + } + + /// Retrieves a page of channel forwarding statistics from the underlying paginated store. + pub fn list_channel_stats( + &self, page_token: Option, + ) -> Result { + let (stats, next_page_token) = + self.runtime.block_on(self.channel_forwarding_stats_store.list_page(page_token))?; + Ok(ChannelForwardingStatsPage { stats, next_page_token }) + } + + /// Retrieves a page of channel pair forwarding statistics from the underlying paginated store. + pub fn list_channel_pair_stats( + &self, page_token: Option, + ) -> Result { + let (stats, next_page_token) = self + .runtime + .block_on(self.channel_pair_forwarding_stats_store.list_page(page_token))?; + Ok(ChannelPairForwardingStatsPage { stats, next_page_token }) + } + + /// Retrieves a page of channel pair forwarding statistics within a specific time range. + /// + /// A bucket matches when its start timestamp is greater than or equal to `start_timestamp` and + /// less than `end_timestamp`. + /// + /// The listing is complete when `next_page_token` is `None`. Matches within each returned page + /// are sorted by bucket start time, but ordering is not global across pages. + pub fn list_channel_pair_stats_in_range( + &self, start_timestamp: u64, end_timestamp: u64, page_token: Option, + ) -> Result { + self.list_channel_pair_stats_filtered_page(page_token, |stats| { + stats.bucket_start_timestamp >= start_timestamp + && stats.bucket_start_timestamp < end_timestamp + }) + } + + /// Retrieves a page of forwarding statistics buckets for a specific channel pair. + /// + /// The listing is complete when `next_page_token` is `None`. Matches within each returned page + /// are sorted by bucket start time, but ordering is not global across pages. + pub fn list_channel_pair_stats_for_pair( + &self, prev_channel_id: ChannelId, next_channel_id: ChannelId, + page_token: Option, + ) -> Result { + self.list_channel_pair_stats_filtered_page(page_token, |stats| { + stats.prev_channel_id == prev_channel_id && stats.next_channel_id == next_channel_id + }) + } +} diff --git a/src/payment/mod.rs b/src/payment/mod.rs index fd75322ceb..f52cddd5e7 100644 --- a/src/payment/mod.rs +++ b/src/payment/mod.rs @@ -10,6 +10,7 @@ pub(crate) mod asynchronous; mod bolt11; mod bolt12; +mod forwarding; mod onchain; pub(crate) mod pending_payment_store; mod spontaneous; @@ -19,11 +20,16 @@ mod unified; pub use bolt11::Bolt11Payment; pub(crate) use bolt11::PaymentMetadata; pub use bolt12::Bolt12Payment; +pub use forwarding::{ + ChannelForwardingStatsPage, ChannelPairForwardingStatsPage, ForwardedPaymentDetailsPage, + Forwarding, +}; pub use onchain::OnchainPayment; pub(crate) use pending_payment_store::{FundingTxCandidate, PendingPaymentDetails}; pub use spontaneous::SpontaneousPayment; pub use store::{ - Channel, ConfirmationStatus, LSPS2Parameters, PaymentDetails, PaymentDirection, PaymentKind, - PaymentStatus, TransactionType, + Channel, ChannelForwardingStats, ChannelPairForwardingStats, ChannelPairStatsId, + ConfirmationStatus, ForwardedPaymentDetails, ForwardedPaymentId, LSPS2Parameters, + PaymentDetails, PaymentDirection, PaymentKind, PaymentStatus, TransactionType, }; pub use unified::{UnifiedPayment, UnifiedPaymentResult}; diff --git a/src/payment/store.rs b/src/payment/store.rs index 9abf8d7422..1c6a3e87ba 100644 --- a/src/payment/store.rs +++ b/src/payment/store.rs @@ -5,8 +5,13 @@ // http://opensource.org/licenses/MIT>, at your option. You may not use this file except in // accordance with one or both of these licenses. +use std::collections::HashMap; +use std::fmt::{Debug, Display}; +use std::sync::Arc; use std::time::{Duration, SystemTime, UNIX_EPOCH}; +use base64::prelude::BASE64_URL_SAFE_NO_PAD; +use base64::Engine; use bitcoin::secp256k1::PublicKey; use bitcoin::{BlockHash, Txid}; use lightning::chain::chaininterface::TransactionType as LdkTransactionType; @@ -14,6 +19,7 @@ use lightning::ln::channelmanager::PaymentId; use lightning::ln::msgs::DecodeError; use lightning::ln::types::ChannelId; use lightning::offers::offer::OfferId; +use lightning::util::logger::Logger as _; use lightning::util::ser::{Readable, Writeable}; use lightning::{ _init_and_read_len_prefixed_tlv_fields, impl_writeable_tlv_based, @@ -22,8 +28,11 @@ use lightning::{ use lightning_types::payment::{PaymentHash, PaymentPreimage, PaymentSecret}; use lightning_types::string::UntrustedString; -use crate::data_store::{StorableObject, StorableObjectId, StorableObjectUpdate}; +use crate::data_store::{DataStore, StorableObject, StorableObjectId, StorableObjectUpdate}; use crate::hex_utils; +use crate::logger::{log_debug, log_error, Logger}; +use crate::Error; +use crate::UserChannelId; /// Represents a payment. #[derive(Clone, Debug, PartialEq, Eq)] @@ -1085,3 +1094,1173 @@ mod tests { assert_eq!(decoded, PaymentKind::read(&mut &*reencoded).unwrap()); } } + +/// A unique identifier for a forwarded payment. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub struct ForwardedPaymentId(pub [u8; 32]); + +impl StorableObjectId for ForwardedPaymentId { + fn encode_to_key(&self) -> String { + hex_utils::to_string(&self.0) + } +} + +impl Writeable for ForwardedPaymentId { + fn write( + &self, writer: &mut W, + ) -> Result<(), lightning::io::Error> { + self.0.write(writer) + } +} + +impl Readable for ForwardedPaymentId { + fn read(reader: &mut R) -> Result { + Ok(Self(Readable::read(reader)?)) + } +} + +/// Details of a payment that has been forwarded through this node. +/// +/// For multi-HTLC forwards with a single incoming or outgoing HTLC, one record is stored for each +/// channel pair that carried part of the forwarded amount. Amounts and fees are divided between +/// the records using the amounts reported for each incoming and outgoing HTLC. Detailed records +/// are omitted when both sides contain multiple HTLCs because LDK does not identify the exact +/// incoming/outgoing pairs. +#[derive(Clone, Debug, PartialEq, Eq)] +#[cfg_attr(feature = "uniffi", derive(uniffi::Record))] +pub struct ForwardedPaymentDetails { + /// A randomly generated identifier for this forwarded payment. + pub id: ForwardedPaymentId, + /// The incoming channel id. + pub prev_channel_id: ChannelId, + /// The outgoing channel id. + pub next_channel_id: ChannelId, + /// The incoming user channel id, if available. + pub prev_user_channel_id: Option, + /// The outgoing user channel id, if available. + pub next_user_channel_id: Option, + /// The previous node id, if available. + pub prev_node_id: Option, + /// The next node id, if available. + pub next_node_id: Option, + /// The inbound amount attributed to this channel pair, in millisatoshis. + pub inbound_amount_forwarded_msat: Option, + /// The fee attributed to this channel pair, in millisatoshis. + pub total_fee_earned_msat: Option, + /// The skimmed fee attributed to this channel pair, in millisatoshis. + pub skimmed_fee_msat: Option, + /// Whether the forwarded HTLC was claimed from an on-chain transaction. + pub claim_from_onchain_tx: bool, + /// The outbound amount attributed to this channel pair, in millisatoshis. + pub outbound_amount_forwarded_msat: Option, + /// The timestamp when this payment was forwarded. + pub forwarded_at_timestamp: u64, +} + +impl_writeable_tlv_based!(ForwardedPaymentDetails, { + (0, id, required), + (2, prev_channel_id, required), + (4, next_channel_id, required), + (6, prev_user_channel_id, option), + (8, next_user_channel_id, option), + (10, prev_node_id, option), + (12, next_node_id, option), + (14, total_fee_earned_msat, option), + (16, skimmed_fee_msat, option), + (18, claim_from_onchain_tx, required), + (20, outbound_amount_forwarded_msat, option), + (22, forwarded_at_timestamp, required), + (24, inbound_amount_forwarded_msat, option), +}); + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub(crate) struct ForwardedPaymentDetailsUpdate { + id: ForwardedPaymentId, +} + +impl StorableObjectUpdate for ForwardedPaymentDetailsUpdate { + fn id(&self) -> ForwardedPaymentId { + self.id + } +} + +impl StorableObject for ForwardedPaymentDetails { + type Id = ForwardedPaymentId; + type Update = ForwardedPaymentDetailsUpdate; + + fn id(&self) -> Self::Id { + self.id + } + + fn update(&mut self, _update: Self::Update) -> bool { + false + } + + fn to_update(&self) -> Self::Update { + ForwardedPaymentDetailsUpdate { id: self.id } + } +} + +/// Aggregate statistics for forwarded payments through a single channel. +#[derive(Clone, Debug, PartialEq, Eq)] +#[cfg_attr(feature = "uniffi", derive(uniffi::Record))] +pub struct ChannelForwardingStats { + /// The channel id these stats apply to. + pub channel_id: ChannelId, + /// The channel counterparty node id, if known. + pub counterparty_node_id: Option, + /// Number of forwarded payments where this was the incoming channel. + pub inbound_payments_forwarded: u64, + /// Number of forwarded payments where this was the outgoing channel. + pub outbound_payments_forwarded: u64, + /// Total inbound amount forwarded through this channel, in millisatoshis. + pub total_inbound_amount_msat: u64, + /// Total outbound amount forwarded through this channel, in millisatoshis. + pub total_outbound_amount_msat: u64, + /// Total forwarding fees earned through this channel, in millisatoshis, if known for every + /// recorded forward. + pub total_fee_earned_msat: Option, + /// Total skimmed fees for this channel, in millisatoshis. + pub total_skimmed_fee_msat: u64, + /// Number of forwarded HTLCs claimed from on-chain transactions. + pub onchain_claims_count: u64, + /// Timestamp of the first forward recorded for this channel. + pub first_forwarded_at_timestamp: u64, + /// Timestamp of the latest forward recorded for this channel. + pub last_forwarded_at_timestamp: u64, +} + +impl_writeable_tlv_based!(ChannelForwardingStats, { + (0, channel_id, required), + (2, counterparty_node_id, option), + (4, inbound_payments_forwarded, required), + (6, outbound_payments_forwarded, required), + (8, total_inbound_amount_msat, required), + (10, total_outbound_amount_msat, required), + (12, total_fee_earned_msat, option), + (14, total_skimmed_fee_msat, required), + (16, onchain_claims_count, required), + (18, first_forwarded_at_timestamp, required), + (20, last_forwarded_at_timestamp, required), +}); + +/// An opaque identifier for a channel-pair statistics bucket, formed from previous channel, next +/// channel, and time bucket. +/// +/// The components identified by this id are exposed as fields on +/// [`ChannelPairForwardingStats`]. +#[derive(Clone, Copy, PartialEq, Eq, Hash)] +pub struct ChannelPairStatsId([u8; 80]); + +impl ChannelPairStatsId { + /// Constructs an identifier from a channel pair, bucket start time, and bucket size. + pub(crate) fn from_channel_pair_and_bucket( + prev: &ChannelId, next: &ChannelId, bucket_start_timestamp: u64, bucket_size_secs: u64, + ) -> Self { + let mut result = [0u8; 80]; + result[0..32].copy_from_slice(&prev.0); + result[32..64].copy_from_slice(&next.0); + result[64..72].copy_from_slice(&bucket_start_timestamp.to_be_bytes()); + result[72..80].copy_from_slice(&bucket_size_secs.to_be_bytes()); + Self(result) + } + + #[cfg(feature = "uniffi")] + pub(crate) fn from_bytes(bytes: [u8; 80]) -> Self { + Self(bytes) + } + + #[cfg(feature = "uniffi")] + pub(crate) fn as_bytes(&self) -> &[u8; 80] { + &self.0 + } +} + +impl Writeable for ChannelPairStatsId { + fn write( + &self, writer: &mut W, + ) -> Result<(), lightning::io::Error> { + writer.write_all(&self.0) + } +} + +impl Readable for ChannelPairStatsId { + fn read(reader: &mut R) -> Result { + let mut bytes = [0u8; 80]; + reader.read_exact(&mut bytes)?; + Ok(Self(bytes)) + } +} + +impl Display for ChannelPairStatsId { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{}", hex_utils::to_string(&self.0)) + } +} + +impl Debug for ChannelPairStatsId { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "ChannelPairStatsId({})", hex_utils::to_string(&self.0)) + } +} + +impl StorableObjectId for ChannelPairStatsId { + fn encode_to_key(&self) -> String { + // The full ID is 160 hex characters, exceeding LDK's 120-character KVStore key limit. + BASE64_URL_SAFE_NO_PAD.encode(self.0) + } +} + +/// Aggregated statistics for a specific channel pair. +#[derive(Clone, Debug, PartialEq, Eq)] +#[cfg_attr(feature = "uniffi", derive(uniffi::Record))] +pub struct ChannelPairForwardingStats { + /// The unique channel-pair bucket id. + pub id: ChannelPairStatsId, + /// The incoming channel id. + pub prev_channel_id: ChannelId, + /// The outgoing channel id. + pub next_channel_id: ChannelId, + /// Start timestamp of this aggregation bucket. + pub bucket_start_timestamp: u64, + /// Width of this aggregation bucket, in seconds. + pub bucket_size_secs: u64, + /// The previous node id, if available. + pub prev_node_id: Option, + /// The next node id, if available. + pub next_node_id: Option, + /// Number of payments aggregated in this bucket. + pub payment_count: u64, + /// Total inbound amount in this bucket, in millisatoshis. + pub total_inbound_amount_msat: u64, + /// Total outbound amount in this bucket, in millisatoshis. + pub total_outbound_amount_msat: u64, + /// Total forwarding fees earned in this bucket, in millisatoshis, if known for every payment. + pub total_fee_earned_msat: Option, + /// Total skimmed fees in this bucket, in millisatoshis. + pub total_skimmed_fee_msat: u64, + /// Number of forwarded HTLCs claimed from on-chain transactions. + pub onchain_claims_count: u64, + /// Average forwarding fee per payment, in millisatoshis, if known for every payment. + pub avg_fee_msat: Option, + /// Average inbound amount per payment, in millisatoshis. + pub avg_inbound_amount_msat: u64, + /// Timestamp of the first forward in this bucket. + pub first_forwarded_at_timestamp: u64, + /// Timestamp of the latest forward in this bucket. + pub last_forwarded_at_timestamp: u64, + /// Timestamp when this bucket was aggregated. + pub aggregated_at_timestamp: u64, +} + +impl_writeable_tlv_based!(ChannelPairForwardingStats, { + (0, id, required), + (2, prev_channel_id, required), + (4, next_channel_id, required), + (6, prev_node_id, option), + (8, next_node_id, option), + (10, payment_count, required), + (12, total_inbound_amount_msat, required), + (14, total_outbound_amount_msat, required), + (16, total_fee_earned_msat, option), + (18, total_skimmed_fee_msat, required), + (20, onchain_claims_count, required), + (22, avg_fee_msat, option), + (24, avg_inbound_amount_msat, required), + (26, first_forwarded_at_timestamp, required), + (28, last_forwarded_at_timestamp, required), + (30, aggregated_at_timestamp, required), + (32, bucket_start_timestamp, required), + (34, bucket_size_secs, required), +}); + +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) struct ChannelForwardingStatsUpdate { + pub channel_id: ChannelId, + pub counterparty_node_id: Option, + pub inbound_payments_increment: u64, + pub outbound_payments_increment: u64, + pub inbound_amount_increment_msat: u64, + pub outbound_amount_increment_msat: u64, + pub fee_earned_increment_msat: Option, + pub skimmed_fee_increment_msat: u64, + pub onchain_claims_increment: u64, + pub timestamp: u64, +} + +impl StorableObjectUpdate for ChannelForwardingStatsUpdate { + fn id(&self) -> ChannelId { + self.channel_id + } +} + +impl StorableObjectId for ChannelId { + fn encode_to_key(&self) -> String { + hex_utils::to_string(&self.0) + } +} + +impl StorableObject for ChannelForwardingStats { + type Id = ChannelId; + type Update = ChannelForwardingStatsUpdate; + + fn id(&self) -> Self::Id { + self.channel_id + } + + fn update(&mut self, update: Self::Update) -> bool { + debug_assert_eq!(self.channel_id, update.channel_id); + let mut updated = false; + if self.counterparty_node_id.is_none() && update.counterparty_node_id.is_some() { + self.counterparty_node_id = update.counterparty_node_id; + updated = true; + } + if update.inbound_payments_increment > 0 { + self.inbound_payments_forwarded += update.inbound_payments_increment; + updated = true; + } + if update.outbound_payments_increment > 0 { + self.outbound_payments_forwarded += update.outbound_payments_increment; + updated = true; + } + if update.inbound_amount_increment_msat > 0 { + self.total_inbound_amount_msat += update.inbound_amount_increment_msat; + updated = true; + } + if update.outbound_amount_increment_msat > 0 { + self.total_outbound_amount_msat += update.outbound_amount_increment_msat; + updated = true; + } + match (self.total_fee_earned_msat.as_mut(), update.fee_earned_increment_msat) { + (Some(total), Some(increment)) if increment > 0 => { + *total += increment; + updated = true; + }, + (Some(_), None) => { + self.total_fee_earned_msat = None; + updated = true; + }, + _ => {}, + } + if update.skimmed_fee_increment_msat > 0 { + self.total_skimmed_fee_msat += update.skimmed_fee_increment_msat; + updated = true; + } + if update.onchain_claims_increment > 0 { + self.onchain_claims_count += update.onchain_claims_increment; + updated = true; + } + if updated { + self.first_forwarded_at_timestamp = + self.first_forwarded_at_timestamp.min(update.timestamp); + self.last_forwarded_at_timestamp = + self.last_forwarded_at_timestamp.max(update.timestamp); + } + updated + } + + fn to_update(&self) -> Self::Update { + ChannelForwardingStatsUpdate { + channel_id: self.channel_id, + counterparty_node_id: self.counterparty_node_id, + inbound_payments_increment: self.inbound_payments_forwarded, + outbound_payments_increment: self.outbound_payments_forwarded, + inbound_amount_increment_msat: self.total_inbound_amount_msat, + outbound_amount_increment_msat: self.total_outbound_amount_msat, + fee_earned_increment_msat: self.total_fee_earned_msat, + skimmed_fee_increment_msat: self.total_skimmed_fee_msat, + onchain_claims_increment: self.onchain_claims_count, + timestamp: self.last_forwarded_at_timestamp, + } + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) struct ChannelPairForwardingStatsUpdate { + pub id: ChannelPairStatsId, + pub prev_node_id: Option, + pub next_node_id: Option, + pub payment_count_increment: u64, + pub inbound_amount_increment_msat: u64, + pub outbound_amount_increment_msat: u64, + pub fee_earned_increment_msat: Option, + pub skimmed_fee_increment_msat: u64, + pub onchain_claims_increment: u64, + pub first_timestamp: u64, + pub last_timestamp: u64, + pub aggregated_at_timestamp: u64, +} + +impl StorableObjectUpdate for ChannelPairForwardingStatsUpdate { + fn id(&self) -> ChannelPairStatsId { + self.id + } +} + +impl StorableObject for ChannelPairForwardingStats { + type Id = ChannelPairStatsId; + type Update = ChannelPairForwardingStatsUpdate; + + fn id(&self) -> Self::Id { + self.id + } + + fn update(&mut self, update: Self::Update) -> bool { + debug_assert_eq!(self.id, update.id); + let mut updated = false; + if self.prev_node_id.is_none() && update.prev_node_id.is_some() { + self.prev_node_id = update.prev_node_id; + updated = true; + } + if self.next_node_id.is_none() && update.next_node_id.is_some() { + self.next_node_id = update.next_node_id; + updated = true; + } + if update.payment_count_increment > 0 { + self.payment_count += update.payment_count_increment; + updated = true; + } + if update.inbound_amount_increment_msat > 0 { + self.total_inbound_amount_msat += update.inbound_amount_increment_msat; + updated = true; + } + if update.outbound_amount_increment_msat > 0 { + self.total_outbound_amount_msat += update.outbound_amount_increment_msat; + updated = true; + } + match (self.total_fee_earned_msat.as_mut(), update.fee_earned_increment_msat) { + (Some(total), Some(increment)) if increment > 0 => { + *total += increment; + updated = true; + }, + (Some(_), None) => { + self.total_fee_earned_msat = None; + updated = true; + }, + _ => {}, + } + if update.skimmed_fee_increment_msat > 0 { + self.total_skimmed_fee_msat += update.skimmed_fee_increment_msat; + updated = true; + } + if update.onchain_claims_increment > 0 { + self.onchain_claims_count += update.onchain_claims_increment; + updated = true; + } + if updated { + if self.first_forwarded_at_timestamp == 0 { + self.first_forwarded_at_timestamp = update.first_timestamp; + } else { + self.first_forwarded_at_timestamp = + self.first_forwarded_at_timestamp.min(update.first_timestamp); + } + self.last_forwarded_at_timestamp = + self.last_forwarded_at_timestamp.max(update.last_timestamp); + if self.payment_count > 0 { + self.avg_fee_msat = + self.total_fee_earned_msat.map(|total| total / self.payment_count); + self.avg_inbound_amount_msat = self.total_inbound_amount_msat / self.payment_count; + } + } + if update.aggregated_at_timestamp > self.aggregated_at_timestamp { + self.aggregated_at_timestamp = update.aggregated_at_timestamp; + updated = true; + } + updated + } + + fn to_update(&self) -> Self::Update { + ChannelPairForwardingStatsUpdate { + id: self.id, + prev_node_id: self.prev_node_id, + next_node_id: self.next_node_id, + payment_count_increment: self.payment_count, + inbound_amount_increment_msat: self.total_inbound_amount_msat, + outbound_amount_increment_msat: self.total_outbound_amount_msat, + fee_earned_increment_msat: self.total_fee_earned_msat, + skimmed_fee_increment_msat: self.total_skimmed_fee_msat, + onchain_claims_increment: self.onchain_claims_count, + first_timestamp: self.first_forwarded_at_timestamp, + last_timestamp: self.last_forwarded_at_timestamp, + aggregated_at_timestamp: self.aggregated_at_timestamp, + } + } +} + +fn seconds_until_next_forwarding_aggregation(now_timestamp: u64, bucket_size_secs: u64) -> u64 { + debug_assert!(bucket_size_secs > 0); + bucket_size_secs - (now_timestamp % bucket_size_secs) +} + +pub(crate) async fn run_forwarded_payment_aggregation( + mut stop_receiver: tokio::sync::watch::Receiver<()>, + forwarded_payment_store: Arc>>, + channel_pair_stats_store: Arc>>, + bucket_size_secs: u64, logger: Arc, +) { + let period = Duration::from_secs(bucket_size_secs); + let now = + SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or(Duration::from_secs(0)).as_secs(); + let secs_until_next_bucket = seconds_until_next_forwarding_aggregation(now, bucket_size_secs); + let first_tick = tokio::time::Instant::now() + Duration::from_secs(secs_until_next_bucket); + let mut interval = tokio::time::interval_at(first_tick, period); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + loop { + tokio::select! { + _ = stop_receiver.changed() => break, + _ = interval.tick() => { + match aggregate_expired_forwarded_payments( + &forwarded_payment_store, + &channel_pair_stats_store, + bucket_size_secs, + &logger, + ) + .await + { + Ok((pair_count, payment_count)) if pair_count > 0 => { + log_debug!( + logger, + "Aggregated {} forwarded payments into {} channel pair buckets", + payment_count, + pair_count + ); + }, + Ok((0, payment_count)) if payment_count > 0 => { + log_debug!( + logger, + "Removed {} forwarded payment details from previously aggregated buckets", + payment_count + ); + }, + Err(e) => log_error!( + logger, + "Periodic forwarded payment aggregation failed: {}", + e + ), + _ => {}, + } + } + } + } +} + +/// Aggregate forwarded payments older than the current and previous buckets into channel-pair +/// statistics. +pub(crate) async fn aggregate_expired_forwarded_payments( + forwarded_payment_store: &DataStore>, + channel_pair_stats_store: &DataStore>, + bucket_size_secs: u64, logger: &Arc, +) -> Result<(u64, u64), Error> { + let now = + SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or(Duration::from_secs(0)).as_secs(); + aggregate_expired_forwarded_payments_at( + forwarded_payment_store, + channel_pair_stats_store, + bucket_size_secs, + now, + logger, + ) + .await +} + +async fn aggregate_expired_forwarded_payments_at( + forwarded_payment_store: &DataStore>, + channel_pair_stats_store: &DataStore>, + bucket_size_secs: u64, now: u64, logger: &Arc, +) -> Result<(u64, u64), Error> { + if bucket_size_secs == 0 { + return Ok((0, 0)); + } + let current_bucket_start = (now / bucket_size_secs).saturating_mul(bucket_size_secs); + let oldest_retained_bucket_start = current_bucket_start.saturating_sub(bucket_size_secs); + + // Retain details from the current and immediately previous buckets. Once an older bucket is + // persisted, it is a durable commit marker: retries can skip updating its totals and finish + // deleting detail records left behind by an interrupted cleanup. + // Detail insertions assign their timestamps while holding this same lock. Holding it until all + // closed-bucket markers are persisted ensures a record can either be included in the marker or + // receive a timestamp after the aggregation pass, but cannot be inserted late into a committed + // bucket. + let forwarded_payment_guard = forwarded_payment_store.mutation_guard().await; + let expired = forwarded_payment_store + .list_filter(|p| p.forwarded_at_timestamp < oldest_retained_bucket_start); + if expired.is_empty() { + log_debug!(logger, "No forwarded payments in closed aggregation buckets found"); + return Ok((0, 0)); + } + + let mut bucket_groups: HashMap<(ChannelId, ChannelId, u64), Vec> = + HashMap::new(); + for payment in expired { + let bucket_start = (payment.forwarded_at_timestamp / bucket_size_secs) * bucket_size_secs; + bucket_groups + .entry((payment.prev_channel_id, payment.next_channel_id, bucket_start)) + .or_default() + .push(payment); + } + + let mut aggregated_bucket_count = 0u64; + let mut payment_ids_to_remove = Vec::new(); + for ((prev_channel_id, next_channel_id, bucket_start), payments) in bucket_groups { + let pair_id = ChannelPairStatsId::from_channel_pair_and_bucket( + &prev_channel_id, + &next_channel_id, + bucket_start, + bucket_size_secs, + ); + + if !channel_pair_stats_store.contains_key(&pair_id) { + let mut total_inbound_amount_msat = 0u64; + let mut total_outbound_amount_msat = 0u64; + let mut total_fee_earned_msat = Some(0u64); + let mut total_skimmed_fee_msat = 0u64; + let mut onchain_claims_count = 0u64; + let mut first_timestamp = u64::MAX; + let mut last_timestamp = 0u64; + + for payment in &payments { + let outbound = payment.outbound_amount_forwarded_msat.unwrap_or(0); + let fee = payment.total_fee_earned_msat; + let skimmed = payment.skimmed_fee_msat.unwrap_or(0); + let inbound = payment + .inbound_amount_forwarded_msat + .unwrap_or_else(|| outbound.saturating_add(fee.unwrap_or(0))); + total_inbound_amount_msat = total_inbound_amount_msat.saturating_add(inbound); + total_outbound_amount_msat = total_outbound_amount_msat.saturating_add(outbound); + total_fee_earned_msat = match (total_fee_earned_msat, fee) { + (Some(total), Some(fee)) => Some(total.saturating_add(fee)), + _ => None, + }; + total_skimmed_fee_msat = total_skimmed_fee_msat.saturating_add(skimmed); + if payment.claim_from_onchain_tx { + onchain_claims_count += 1; + } + first_timestamp = first_timestamp.min(payment.forwarded_at_timestamp); + last_timestamp = last_timestamp.max(payment.forwarded_at_timestamp); + } + + let payment_count = payments.len() as u64; + let prev_node_id = payments.iter().find_map(|payment| payment.prev_node_id); + let next_node_id = payments.iter().find_map(|payment| payment.next_node_id); + let stats = ChannelPairForwardingStats { + id: pair_id, + prev_channel_id, + next_channel_id, + bucket_start_timestamp: bucket_start, + bucket_size_secs, + prev_node_id, + next_node_id, + payment_count, + total_inbound_amount_msat, + total_outbound_amount_msat, + total_fee_earned_msat, + total_skimmed_fee_msat, + onchain_claims_count, + avg_fee_msat: total_fee_earned_msat.map(|total| total / payment_count), + avg_inbound_amount_msat: total_inbound_amount_msat / payment_count, + first_forwarded_at_timestamp: first_timestamp, + last_forwarded_at_timestamp: last_timestamp, + aggregated_at_timestamp: now, + }; + + channel_pair_stats_store.insert(stats).await.map_err(|e| { + log_error!( + logger, + "Failed to insert channel pair stats bucket for {pair_id:?}: {e}" + ); + e + })?; + aggregated_bucket_count += 1; + } + + payment_ids_to_remove.extend(payments.into_iter().map(|payment| payment.id)); + } + + // Removals acquire the mutation lock themselves. Once every bucket marker has been persisted, + // detail insertions may resume without risking insertion into any bucket closed by this pass. + drop(forwarded_payment_guard); + let mut removed_payment_count = 0u64; + for payment_id in payment_ids_to_remove { + forwarded_payment_store.remove(&payment_id).await.map_err(|e| { + log_error!(logger, "Failed to remove forwarded payment {:?}: {}", payment_id, e); + e + })?; + removed_payment_count += 1; + } + + Ok((aggregated_bucket_count, removed_payment_count)) +} + +/// Aggregates multiple channel-pair statistics buckets into cumulative totals. +/// +/// The returned bucket spans from the earliest input bucket start through the latest input bucket +/// end, including any gaps between buckets. +/// +/// Returns `None` if `buckets` is empty or contains statistics for different channel pairs. +pub fn aggregate_channel_pair_stats( + buckets: &[ChannelPairForwardingStats], +) -> Option { + let first = buckets.first()?; + for bucket in &buckets[1..] { + if bucket.prev_channel_id != first.prev_channel_id + || bucket.next_channel_id != first.next_channel_id + { + return None; + } + } + + let mut payment_count = 0u64; + let mut total_inbound_amount_msat = 0u64; + let mut total_outbound_amount_msat = 0u64; + let mut total_fee_earned_msat = Some(0u64); + let mut total_skimmed_fee_msat = 0u64; + let mut onchain_claims_count = 0u64; + let mut first_forwarded_at_timestamp = u64::MAX; + let mut last_forwarded_at_timestamp = 0u64; + let mut earliest_bucket_start = u64::MAX; + let mut latest_bucket_end = 0u64; + let mut prev_node_id = None; + let mut next_node_id = None; + for bucket in buckets { + payment_count = payment_count.saturating_add(bucket.payment_count); + total_inbound_amount_msat = + total_inbound_amount_msat.saturating_add(bucket.total_inbound_amount_msat); + total_outbound_amount_msat = + total_outbound_amount_msat.saturating_add(bucket.total_outbound_amount_msat); + total_fee_earned_msat = match (total_fee_earned_msat, bucket.total_fee_earned_msat) { + (Some(total), Some(fee)) => Some(total.saturating_add(fee)), + _ => None, + }; + total_skimmed_fee_msat = + total_skimmed_fee_msat.saturating_add(bucket.total_skimmed_fee_msat); + onchain_claims_count = onchain_claims_count.saturating_add(bucket.onchain_claims_count); + first_forwarded_at_timestamp = + first_forwarded_at_timestamp.min(bucket.first_forwarded_at_timestamp); + last_forwarded_at_timestamp = + last_forwarded_at_timestamp.max(bucket.last_forwarded_at_timestamp); + earliest_bucket_start = earliest_bucket_start.min(bucket.bucket_start_timestamp); + latest_bucket_end = latest_bucket_end + .max(bucket.bucket_start_timestamp.saturating_add(bucket.bucket_size_secs)); + if prev_node_id.is_none() { + prev_node_id = bucket.prev_node_id; + } + if next_node_id.is_none() { + next_node_id = bucket.next_node_id; + } + } + let bucket_size_secs = latest_bucket_end.saturating_sub(earliest_bucket_start); + let now = + SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or(Duration::from_secs(0)).as_secs(); + Some(ChannelPairForwardingStats { + id: ChannelPairStatsId::from_channel_pair_and_bucket( + &first.prev_channel_id, + &first.next_channel_id, + earliest_bucket_start, + bucket_size_secs, + ), + prev_channel_id: first.prev_channel_id, + next_channel_id: first.next_channel_id, + bucket_start_timestamp: earliest_bucket_start, + bucket_size_secs, + prev_node_id, + next_node_id, + payment_count, + total_inbound_amount_msat, + total_outbound_amount_msat, + total_fee_earned_msat, + total_skimmed_fee_msat, + onchain_claims_count, + avg_fee_msat: if payment_count > 0 { + total_fee_earned_msat.map(|total| total / payment_count) + } else { + None + }, + avg_inbound_amount_msat: if payment_count > 0 { + total_inbound_amount_msat / payment_count + } else { + 0 + }, + first_forwarded_at_timestamp, + last_forwarded_at_timestamp, + aggregated_at_timestamp: now, + }) +} + +#[cfg(test)] +mod forwarding_stats_tests { + use std::str::FromStr; + + use lightning::util::persist::{ + KVStore, KVSTORE_NAMESPACE_KEY_ALPHABET, KVSTORE_NAMESPACE_KEY_MAX_LEN, + }; + + use super::*; + use crate::io::sqlite_store::SqliteStore; + use crate::io::test_utils::{random_storage_path, InMemoryStore}; + use crate::types::{DynStore, DynStoreWrapper}; + + type TestForwardedPaymentStore = DataStore>; + type TestChannelPairStatsStore = DataStore>; + + fn test_stores() -> (TestForwardedPaymentStore, TestChannelPairStatsStore, Arc) { + let kv_store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let logger = Arc::new(Logger::new_log_facade()); + let forwarded_payment_store = DataStore::new( + Vec::new(), + "test_forwarded_payments".to_string(), + String::new(), + Arc::clone(&kv_store), + Arc::clone(&logger), + ); + let channel_pair_stats_store = DataStore::new( + Vec::new(), + "test_channel_pair_stats".to_string(), + String::new(), + kv_store, + Arc::clone(&logger), + ); + (forwarded_payment_store, channel_pair_stats_store, logger) + } + + fn forwarded_payment( + id: u8, forwarded_at_timestamp: u64, inbound_amount_msat: u64, outbound_amount_msat: u64, + fee_msat: u64, + ) -> ForwardedPaymentDetails { + ForwardedPaymentDetails { + id: ForwardedPaymentId([id; 32]), + prev_channel_id: ChannelId([1; 32]), + next_channel_id: ChannelId([2; 32]), + prev_user_channel_id: None, + next_user_channel_id: None, + prev_node_id: None, + next_node_id: None, + inbound_amount_forwarded_msat: Some(inbound_amount_msat), + total_fee_earned_msat: Some(fee_msat), + skimmed_fee_msat: Some(0), + claim_from_onchain_tx: false, + outbound_amount_forwarded_msat: Some(outbound_amount_msat), + forwarded_at_timestamp, + } + } + + fn channel_pair_stats( + bucket_start_timestamp: u64, bucket_size_secs: u64, aggregated_at_timestamp: u64, + payment_count: u64, total_inbound_amount_msat: u64, total_outbound_amount_msat: u64, + total_fee_earned_msat: u64, first_forwarded_at_timestamp: u64, + last_forwarded_at_timestamp: u64, + ) -> ChannelPairForwardingStats { + let prev_channel_id = ChannelId([1; 32]); + let next_channel_id = ChannelId([2; 32]); + ChannelPairForwardingStats { + id: ChannelPairStatsId::from_channel_pair_and_bucket( + &prev_channel_id, + &next_channel_id, + bucket_start_timestamp, + bucket_size_secs, + ), + prev_channel_id, + next_channel_id, + bucket_start_timestamp, + bucket_size_secs, + prev_node_id: None, + next_node_id: None, + payment_count, + total_inbound_amount_msat, + total_outbound_amount_msat, + total_fee_earned_msat: Some(total_fee_earned_msat), + total_skimmed_fee_msat: 0, + onchain_claims_count: 0, + avg_fee_msat: Some(total_fee_earned_msat / payment_count), + avg_inbound_amount_msat: total_inbound_amount_msat / payment_count, + first_forwarded_at_timestamp, + last_forwarded_at_timestamp, + aggregated_at_timestamp, + } + } + + #[test] + fn channel_pair_persistence_key_fits_kvstore_limit() { + let id = ChannelPairStatsId::from_channel_pair_and_bucket( + &ChannelId([1; 32]), + &ChannelId([2; 32]), + 42, + 60, + ); + let other_id = ChannelPairStatsId::from_channel_pair_and_bucket( + &ChannelId([1; 32]), + &ChannelId([2; 32]), + 43, + 60, + ); + let other_size_id = ChannelPairStatsId::from_channel_pair_and_bucket( + &ChannelId([1; 32]), + &ChannelId([2; 32]), + 42, + 120, + ); + + let key = id.encode_to_key(); + assert_eq!(key.len(), 107); + assert!(key.len() <= KVSTORE_NAMESPACE_KEY_MAX_LEN); + assert!(key.chars().all(|c| KVSTORE_NAMESPACE_KEY_ALPHABET.contains(c))); + assert_eq!(BASE64_URL_SAFE_NO_PAD.decode(&key).unwrap(), id.0); + assert_ne!(key, other_id.encode_to_key()); + assert_ne!(key, other_size_id.encode_to_key()); + } + + #[tokio::test] + async fn aggregation_retains_current_and_previous_buckets() { + let (forwarded_payment_store, channel_pair_stats_store, logger) = test_stores(); + let closed_bucket_payment = forwarded_payment(1, 899, 110, 100, 10); + let partial_bucket_payment = forwarded_payment(2, 939, 220, 200, 20); + forwarded_payment_store.insert(closed_bucket_payment.clone()).await.unwrap(); + forwarded_payment_store.insert(partial_bucket_payment.clone()).await.unwrap(); + + // At timestamp 1,000, the current bucket starts at 960 and the previous bucket starts at + // 900. Only payments older than the previous bucket are aggregated. + assert_eq!( + aggregate_expired_forwarded_payments_at( + &forwarded_payment_store, + &channel_pair_stats_store, + 60, + 1_000, + &logger, + ) + .await, + Ok((1, 1)) + ); + + assert!(forwarded_payment_store.get(&closed_bucket_payment.id).is_none()); + assert_eq!( + forwarded_payment_store.get(&partial_bucket_payment.id), + Some(partial_bucket_payment) + ); + let bucket_id = ChannelPairStatsId::from_channel_pair_and_bucket( + &ChannelId([1; 32]), + &ChannelId([2; 32]), + 840, + 60, + ); + let stats = channel_pair_stats_store.get(&bucket_id).unwrap(); + assert_eq!(stats.bucket_size_secs, 60); + assert_eq!(stats.payment_count, 1); + assert_eq!(stats.total_inbound_amount_msat, 110); + assert_eq!(stats.total_outbound_amount_msat, 100); + assert_eq!(stats.total_fee_earned_msat, Some(10)); + assert_eq!(stats.aggregated_at_timestamp, 1_000); + } + + #[test] + fn aggregation_schedule_aligns_to_bucket_closure() { + assert_eq!(seconds_until_next_forwarding_aggregation(120, 60), 60); + assert_eq!(seconds_until_next_forwarding_aggregation(121, 60), 59); + assert_eq!(seconds_until_next_forwarding_aggregation(179, 60), 1); + } + + #[tokio::test] + async fn aggregation_preserves_unknown_fees() { + let (forwarded_payment_store, channel_pair_stats_store, logger) = test_stores(); + let known_fee_payment = forwarded_payment(1, 850, 110, 100, 10); + let mut unknown_fee_payment = forwarded_payment(2, 851, 200, 200, 0); + unknown_fee_payment.total_fee_earned_msat = None; + forwarded_payment_store.insert(known_fee_payment).await.unwrap(); + forwarded_payment_store.insert(unknown_fee_payment).await.unwrap(); + + assert_eq!( + aggregate_expired_forwarded_payments_at( + &forwarded_payment_store, + &channel_pair_stats_store, + 60, + 1_000, + &logger, + ) + .await, + Ok((1, 2)) + ); + + let bucket_id = ChannelPairStatsId::from_channel_pair_and_bucket( + &ChannelId([1; 32]), + &ChannelId([2; 32]), + 840, + 60, + ); + let stats = channel_pair_stats_store.get(&bucket_id).unwrap(); + assert_eq!(stats.payment_count, 2); + assert_eq!(stats.total_fee_earned_msat, None); + assert_eq!(stats.avg_fee_msat, None); + } + + #[test] + fn channel_stats_update_preserves_unknown_fees() { + let mut stats = ChannelForwardingStats { + channel_id: ChannelId([1; 32]), + counterparty_node_id: None, + inbound_payments_forwarded: 1, + outbound_payments_forwarded: 0, + total_inbound_amount_msat: 110, + total_outbound_amount_msat: 0, + total_fee_earned_msat: Some(10), + total_skimmed_fee_msat: 0, + onchain_claims_count: 0, + first_forwarded_at_timestamp: 850, + last_forwarded_at_timestamp: 850, + }; + + assert!(stats.update(ChannelForwardingStatsUpdate { + channel_id: stats.channel_id, + counterparty_node_id: None, + inbound_payments_increment: 1, + outbound_payments_increment: 0, + inbound_amount_increment_msat: 200, + outbound_amount_increment_msat: 0, + fee_earned_increment_msat: None, + skimmed_fee_increment_msat: 0, + onchain_claims_increment: 0, + timestamp: 851, + })); + assert_eq!(stats.total_fee_earned_msat, None); + } + + #[tokio::test] + async fn aggregation_keeps_both_retained_bucket_boundaries() { + let (forwarded_payment_store, channel_pair_stats_store, logger) = test_stores(); + let older_bucket_payment = forwarded_payment(1, 839, 110, 100, 10); + let previous_bucket_payment = forwarded_payment(2, 840, 220, 200, 20); + let current_bucket_payment = forwarded_payment(3, 900, 330, 300, 30); + forwarded_payment_store.insert(older_bucket_payment.clone()).await.unwrap(); + forwarded_payment_store.insert(previous_bucket_payment.clone()).await.unwrap(); + forwarded_payment_store.insert(current_bucket_payment.clone()).await.unwrap(); + + assert_eq!( + aggregate_expired_forwarded_payments_at( + &forwarded_payment_store, + &channel_pair_stats_store, + 60, + 900, + &logger, + ) + .await, + Ok((1, 1)) + ); + + assert!(forwarded_payment_store.get(&older_bucket_payment.id).is_none()); + assert_eq!( + forwarded_payment_store.get(&previous_bucket_payment.id), + Some(previous_bucket_payment) + ); + assert_eq!( + forwarded_payment_store.get(¤t_bucket_payment.id), + Some(current_bucket_payment) + ); + } + + #[tokio::test] + async fn aggregation_retry_only_cleans_up_committed_bucket() { + let (forwarded_payment_store, channel_pair_stats_store, logger) = test_stores(); + let remaining_payment = forwarded_payment(2, 851, 220, 200, 20); + let committed_stats = channel_pair_stats(840, 60, 950, 2, 330, 300, 30, 850, 851); + channel_pair_stats_store.insert(committed_stats.clone()).await.unwrap(); + forwarded_payment_store.insert(remaining_payment.clone()).await.unwrap(); + + // This represents a retry after the bucket write and one of two detail deletions + // succeeded. The existing bucket is the commit marker, so its totals must not change. + assert_eq!( + aggregate_expired_forwarded_payments_at( + &forwarded_payment_store, + &channel_pair_stats_store, + 60, + 1_000, + &logger, + ) + .await, + Ok((0, 1)) + ); + + assert!(forwarded_payment_store.get(&remaining_payment.id).is_none()); + assert_eq!(channel_pair_stats_store.get(&committed_stats.id), Some(committed_stats)); + } + + #[test] + fn channel_pair_update_refreshes_aggregation_timestamp() { + let mut stats = channel_pair_stats(840, 60, 950, 2, 330, 300, 30, 850, 851); + let update = channel_pair_stats(840, 60, 1_000, 1, 110, 100, 10, 852, 852); + + assert!(stats.update(update.to_update())); + assert_eq!(stats.payment_count, 3); + assert_eq!(stats.aggregated_at_timestamp, 1_000); + } + + #[test] + fn cumulative_stats_prefer_known_node_ids_and_cover_bucket_span() { + let unknown_nodes = channel_pair_stats(840, 60, 950, 2, 330, 300, 30, 850, 851); + let mut known_nodes = channel_pair_stats(900, 120, 1_000, 1, 110, 100, 10, 902, 902); + let node_id = PublicKey::from_str( + "0279be667ef9dcbbac55a06295ce870b07029bfcdb2dce28d959f2815b16f81798", + ) + .unwrap(); + known_nodes.prev_node_id = Some(node_id); + known_nodes.next_node_id = Some(node_id); + + let cumulative = aggregate_channel_pair_stats(&[unknown_nodes, known_nodes]).unwrap(); + assert_eq!(cumulative.prev_node_id, Some(node_id)); + assert_eq!(cumulative.next_node_id, Some(node_id)); + assert_eq!(cumulative.bucket_start_timestamp, 840); + assert_eq!(cumulative.bucket_size_secs, 180); + } + + #[tokio::test] + async fn configurable_bucket_length_is_part_of_bucket_geometry() { + let (forwarded_payment_store, channel_pair_stats_store, logger) = test_stores(); + let old_stats = channel_pair_stats(0, 3_600, 7_200, 1, 110, 100, 10, 100, 100); + let payment = forwarded_payment(2, 4_000, 220, 200, 20); + channel_pair_stats_store.insert(old_stats.clone()).await.unwrap(); + forwarded_payment_store.insert(payment.clone()).await.unwrap(); + + assert_eq!( + aggregate_expired_forwarded_payments_at( + &forwarded_payment_store, + &channel_pair_stats_store, + 7_200, + 15_000, + &logger, + ) + .await, + Ok((1, 1)) + ); + + let new_id = ChannelPairStatsId::from_channel_pair_and_bucket( + &payment.prev_channel_id, + &payment.next_channel_id, + 0, + 7_200, + ); + assert_eq!(channel_pair_stats_store.get(&old_stats.id), Some(old_stats)); + assert_eq!(channel_pair_stats_store.get(&new_id).unwrap().payment_count, 1); + assert!(forwarded_payment_store.get(&payment.id).is_none()); + } + + #[tokio::test] + async fn channel_pair_stats_persist_to_sqlite() { + let mut storage_path = random_storage_path(); + storage_path.push("channel_pair_stats_persist_to_sqlite"); + let sqlite_store = + SqliteStore::new(storage_path, Some("stats.sqlite".to_string()), None).unwrap(); + let kv_store: Arc = Arc::new(DynStoreWrapper(sqlite_store)); + let logger = Arc::new(Logger::new_log_facade()); + let namespace = "sqlite_channel_pair_stats"; + let stats_store = DataStore::new( + Vec::new(), + namespace.to_string(), + String::new(), + Arc::clone(&kv_store), + logger, + ); + let stats = channel_pair_stats(840, 60, 1_000, 2, 330, 300, 30, 850, 851); + + stats_store.insert(stats.clone()).await.unwrap(); + let keys = KVStore::list(&*kv_store, namespace, "").await.unwrap(); + assert_eq!(keys, vec![stats.id.encode_to_key()]); + let bytes = KVStore::read(&*kv_store, namespace, "", &keys[0]).await.unwrap(); + assert_eq!(ChannelPairForwardingStats::read(&mut &bytes[..]).unwrap(), stats); + } +} diff --git a/src/types.rs b/src/types.rs index e24db4d253..347392fcf1 100644 --- a/src/types.rs +++ b/src/types.rs @@ -48,6 +48,9 @@ use crate::fee_estimator::OnchainFeeEstimator; use crate::ffi::maybe_wrap; use crate::logger::Logger; use crate::message_handler::NodeCustomMessageHandler; +use crate::payment::store::{ + ChannelForwardingStats, ChannelPairForwardingStats, ForwardedPaymentDetails, +}; use crate::payment::{PaymentDetails, PendingPaymentDetails}; use crate::runtime::RuntimeSpawner; @@ -372,6 +375,10 @@ pub(crate) type BumpTransactionEventHandler = >; pub(crate) type PaymentStore = DataStore>; +pub(crate) type ForwardedPaymentStore = DataStore>; +pub(crate) type ChannelForwardingStatsStore = DataStore>; +pub(crate) type ChannelPairForwardingStatsStore = + DataStore>; /// A local, potentially user-provided, identifier of a channel. ///