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/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 b1ed816df9..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 { @@ -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 { @@ -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"); @@ -115,7 +126,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, @@ -179,13 +190,79 @@ 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 } 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> { @@ -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}; @@ -238,7 +317,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 +433,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) @@ -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 91ab7b27de..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, @@ -1451,7 +1633,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 +1644,7 @@ where from_prev_str, next_htlcs.len(), to_next_str, - outbound_amount_forwarded_msat.unwrap_or(0), + outbound_amount_forwarded_msat, fee_earned, ); } @@ -1490,13 +1672,185 @@ 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(), 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); @@ -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 38583dd7e7..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)] @@ -149,7 +158,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) } } @@ -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. ///