diff --git a/Cargo.toml b/Cargo.toml index 0110c027b..ebd808c00 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -86,18 +86,18 @@ uniffi-default = [ #lightning-macros = { version = "0.2.0" } #lightning-dns-resolver = { version = "0.3.0" } -lightning = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "9174965af9437196c527a9aa0df36bbcf050c8bb", features = ["std"] } -lightning-types = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "9174965af9437196c527a9aa0df36bbcf050c8bb" } -lightning-invoice = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "9174965af9437196c527a9aa0df36bbcf050c8bb", features = ["std"] } -lightning-net-tokio = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "9174965af9437196c527a9aa0df36bbcf050c8bb" } -lightning-persister = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "9174965af9437196c527a9aa0df36bbcf050c8bb", features = ["tokio"], optional = true } -lightning-background-processor = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "9174965af9437196c527a9aa0df36bbcf050c8bb" } -lightning-rapid-gossip-sync = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "9174965af9437196c527a9aa0df36bbcf050c8bb" } -lightning-block-sync = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "9174965af9437196c527a9aa0df36bbcf050c8bb", features = ["rest-client", "rpc-client", "tokio"], optional = true } -lightning-transaction-sync = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "9174965af9437196c527a9aa0df36bbcf050c8bb", optional = true } -lightning-liquidity = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "9174965af9437196c527a9aa0df36bbcf050c8bb", features = ["std"] } -lightning-macros = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "9174965af9437196c527a9aa0df36bbcf050c8bb" } -lightning-dns-resolver = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "9174965af9437196c527a9aa0df36bbcf050c8bb", optional = true } +lightning = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "22cdec713c9c2f077d4e9f51c8f5cd615300085a", features = ["std"] } +lightning-types = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "22cdec713c9c2f077d4e9f51c8f5cd615300085a" } +lightning-invoice = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "22cdec713c9c2f077d4e9f51c8f5cd615300085a", features = ["std"] } +lightning-net-tokio = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "22cdec713c9c2f077d4e9f51c8f5cd615300085a" } +lightning-persister = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "22cdec713c9c2f077d4e9f51c8f5cd615300085a", features = ["tokio"], optional = true } +lightning-background-processor = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "22cdec713c9c2f077d4e9f51c8f5cd615300085a" } +lightning-rapid-gossip-sync = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "22cdec713c9c2f077d4e9f51c8f5cd615300085a" } +lightning-block-sync = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "22cdec713c9c2f077d4e9f51c8f5cd615300085a", features = ["rest-client", "rpc-client", "tokio"], optional = true } +lightning-transaction-sync = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "22cdec713c9c2f077d4e9f51c8f5cd615300085a", optional = true } +lightning-liquidity = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "22cdec713c9c2f077d4e9f51c8f5cd615300085a", features = ["std"] } +lightning-macros = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "22cdec713c9c2f077d4e9f51c8f5cd615300085a" } +lightning-dns-resolver = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "22cdec713c9c2f077d4e9f51c8f5cd615300085a", optional = true } 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"], optional = true } @@ -132,14 +132,14 @@ postgres-native-tls = { version = "0.5", default-features = false, features = [" vss-client = { package = "vss-client-ng", version = "0.6", optional = true } prost = { version = "0.11.6", default-features = false, optional = true} #bitcoin-payment-instructions = { version = "0.6" } -bitcoin-payment-instructions = { git = "https://github.com/tnull/bitcoin-payment-instructions", rev = "1d6ffaa8962391ddc84aeef98bd7439e55ccae9d", optional = true } +bitcoin-payment-instructions = { git = "https://github.com/benthecarman/bitcoin-payment-instructions", rev = "224b07fcc45a687bd56d8a9bded1e837d3d1bc37", optional = true } [target.'cfg(windows)'.dependencies] winapi = { version = "0.3", features = ["winbase"] } [dev-dependencies] -lightning = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "9174965af9437196c527a9aa0df36bbcf050c8bb", features = ["std", "_test_utils"] } -lightning-persister = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "9174965af9437196c527a9aa0df36bbcf050c8bb", features = ["tokio"] } +lightning = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "22cdec713c9c2f077d4e9f51c8f5cd615300085a", features = ["std", "_test_utils"] } +lightning-persister = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "22cdec713c9c2f077d4e9f51c8f5cd615300085a", 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 4c4c1a438..b2f3d74ac 100644 --- a/bindings/ldk_node.udl +++ b/bindings/ldk_node.udl @@ -10,6 +10,8 @@ typedef dictionary ElectrumSyncConfig; typedef dictionary TorConfig; +typedef enum ForwardedPaymentTrackingMode; + typedef interface NodeEntropy; typedef interface ProbingConfig; @@ -68,6 +70,7 @@ interface Node { SpontaneousPayment spontaneous_payment(); OnchainPayment onchain_payment(); Liquidity liquidity(); + ForwardingAnalytics forwarding_analytics(); [Throws=NodeError] void lnurl_auth(string lnurl); [Throws=NodeError] @@ -139,6 +142,8 @@ interface FeeRate { typedef interface Liquidity; +typedef interface ForwardingAnalytics; + [Error] enum NodeError { "AlreadyRunning", @@ -177,6 +182,8 @@ enum NodeError { "InvalidOfferId", "InvalidNodeId", "InvalidPaymentId", + "InvalidForwardedPaymentId", + "InvalidChannelPairForwardingStatsId", "InvalidPaymentHash", "InvalidPaymentPreimage", "InvalidPaymentSecret", @@ -358,6 +365,12 @@ typedef string OfferId; [Custom] typedef string PaymentId; +[Custom] +typedef string ForwardedPaymentId; + +[Custom] +typedef string ChannelPairForwardingStatsId; + [Custom] typedef string PaymentHash; @@ -395,3 +408,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 ab641cf86..a85afb24c 100644 --- a/src/builder.rs +++ b/src/builder.rs @@ -80,7 +80,9 @@ use crate::io::utils::{ #[cfg(feature = "storage-vss")] 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_SECONDARY_NAMESPACE, + FORWARDED_PAYMENT_PERSISTENCE_PRIMARY_NAMESPACE, PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, + PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, PENDING_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, }; @@ -89,6 +91,7 @@ use crate::lnurl_auth::LnurlAuth; use crate::logger::{log_error, LdkLogger, LogLevel, LogWriter, Logger}; use crate::message_handler::NodeCustomMessageHandler; use crate::payment::asynchronous::om_mailbox::OnionMessageMailbox; +use crate::payment::forwarding_store::ForwardingStore; #[cfg(feature = "unified-payments")] use crate::payment::HRNResolver; use crate::peer_store::PeerStore; @@ -1524,26 +1527,37 @@ 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, address_pool_res) = runtime - .block_on(async move { - tokio::join!( - read_n_objects( - &*kv_store_ref, - PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, - PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, - PAYMENT_CACHE_WARMUP_COUNT, - 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), - ), - read_address_pool(&*kv_store_ref, &*logger_ref) - ) - }); + let ( + payment_store_res, + channel_forwarding_stats_res, + node_metris_res, + pending_payment_store_res, + address_pool_res, + ) = runtime.block_on(async move { + tokio::join!( + read_n_objects( + &*kv_store_ref, + PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, + PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, + PAYMENT_CACHE_WARMUP_COUNT, + Arc::clone(&logger_ref), + ), + read_all_objects( + &*kv_store_ref, + FORWARDED_PAYMENT_PERSISTENCE_PRIMARY_NAMESPACE, + CHANNEL_FORWARDING_STATS_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), + ), + read_address_pool(&*kv_store_ref, &*logger_ref), + ) + }); // Initialize the status fields. let node_metrics = match node_metris_res { @@ -1576,6 +1590,14 @@ fn build_with_store_internal( }, }; + let channel_forwarding_stats = match channel_forwarding_stats_res { + Ok(stats) => stats, + Err(e) => { + log_error!(logger, "Failed to read channel forwarding stats from store: {}", e); + return Err(BuildError::ReadFailed); + }, + }; + let (chain_source, chain_tip_opt) = match chain_data_source_config { #[cfg(feature = "chain-esplora")] Some(ChainDataSourceConfig::Esplora { server_url, headers, sync_config }) => { @@ -1902,6 +1924,12 @@ fn build_with_store_internal( Arc::clone(&wallet), Arc::clone(&logger), )); + let forwarding_store = Arc::new(ForwardingStore::new( + channel_forwarding_stats, + config.forwarded_payment_tracking_mode, + Arc::clone(&kv_store), + Arc::clone(&logger), + )); let peer_storage_key = keys_manager.get_peer_storage_key(); let monitor_reader = Arc::new(AsyncPersister::new( @@ -2457,6 +2485,16 @@ fn build_with_store_internal( _leak_checker.0.push(Arc::downgrade(&wallet) as Weak); } + // How long detail records are kept before being folded into channel-pair buckets. `Stats` keeps + // none of its own, and only drains records a previous `Detailed` configuration left behind. + let forwarded_payment_aggregation_retention_secs = match config.forwarded_payment_tracking_mode + { + crate::config::ForwardedPaymentTrackingMode::Detailed => { + crate::payment::forwarding_store::FORWARDED_PAYMENT_AGGREGATION_BUCKET_SIZE_SECS + }, + crate::config::ForwardedPaymentTrackingMode::Stats => 0, + }; + Ok(Node { runtime, stop_sender, @@ -2484,6 +2522,8 @@ fn build_with_store_internal( scorer, peer_store, payment_store, + forwarding_store, + forwarded_payment_aggregation_retention_secs, lnurl_auth, is_running, node_metrics, diff --git a/src/config.rs b/src/config.rs index 65e256117..ac3b8e6e8 100644 --- a/src/config.rs +++ b/src/config.rs @@ -169,6 +169,30 @@ pub(crate) const LIQUIDITY_DISCOVERY_RETRY_INITIAL_DELAY: Duration = Duration::f // thereafter until every configured LSP has been discovered. pub(crate) const LIQUIDITY_DISCOVERY_RETRY_MAX_DELAY: Duration = Duration::from_secs(60 * 60); +/// The mode used for tracking forwarded payments. +/// +/// In either mode, a forward is tracked only when it has exactly one incoming HTLC and one outgoing +/// HTLC, and LDK reports both the outbound amount and total fee. +#[derive(Copy, Clone, Debug, PartialEq, Eq)] +#[cfg_attr(feature = "uniffi", derive(uniffi::Enum))] +pub enum ForwardedPaymentTrackingMode { + /// Track eligible new forwarded payments only as per-channel aggregate statistics. + /// + /// Any detailed records left by a previous configuration are aggregated and removed after their + /// current one-hour bucket closes. + Stats, + /// Store eligible individual forwarded payments for the current and previous one-hour buckets. + /// + /// Payments from older buckets are aggregated into channel-pair statistics and removed. + Detailed, +} + +impl Default for ForwardedPaymentTrackingMode { + fn default() -> Self { + Self::Stats + } +} + #[derive(Debug, Clone)] #[cfg_attr(feature = "uniffi", derive(uniffi::Record))] /// Represents the configuration of an [`Node`] instance. @@ -192,9 +216,10 @@ pub(crate) const LIQUIDITY_DISCOVERY_RETRY_MAX_DELAY: Duration = Duration::from_ doc = "| `hrn_config` | HumanReadableNamesConfig::default() |" )] /// | `manually_handle_unknown_bolt11_payments` | false | +/// | `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 { @@ -268,6 +293,8 @@ pub struct Config { /// /// [`Event::PaymentClaimable`]: crate::Event::PaymentClaimable pub manually_handle_unknown_bolt11_payments: bool, + /// The mode used for tracking forwarded payments. + pub forwarded_payment_tracking_mode: ForwardedPaymentTrackingMode, } impl Default for Config { @@ -286,6 +313,7 @@ impl Default for Config { #[cfg(feature = "unified-payments")] hrn_config: HumanReadableNamesConfig::default(), manually_handle_unknown_bolt11_payments: false, + forwarded_payment_tracking_mode: ForwardedPaymentTrackingMode::default(), } } } diff --git a/src/data_store.rs b/src/data_store.rs index a9fe0d0f5..bdd190621 100644 --- a/src/data_store.rs +++ b/src/data_store.rs @@ -13,7 +13,7 @@ use std::ops::Deref; use std::sync::{Arc, Mutex}; use lightning::io::ErrorKind; -use lightning::util::persist::{KVStore, PageToken, PaginatedKVStore}; +use lightning::util::persist::{KVStore, PageToken, PaginatedKVStore, PaginatedListResponse}; use lightning::util::ser::{Readable, Writeable}; use crate::io::utils::process_kv_store_reads; @@ -21,11 +21,20 @@ use crate::logger::{log_debug, log_error, LdkLogger}; use crate::types::DynStore; use crate::Error; +/// An object a store can read, write and delete as a whole, keyed by its own id. pub(crate) trait StorableObject: Clone + Readable + Writeable { type Id: StorableObjectId; - type Update: StorableObjectUpdate; fn id(&self) -> Self::Id; +} + +/// A [`StorableObject`] that a [`DataStore`] can merge an update into in place. +/// +/// Separate from [`StorableObject`] because stores that only ever replace whole objects have no use +/// for this, and requiring it of them would mean supplying an update representation nothing calls. +pub(crate) trait UpdatableObject: StorableObject { + type Update: StorableObjectUpdate; + fn update(&mut self, update: Self::Update) -> bool; fn to_update(&self) -> Self::Update; } @@ -55,6 +64,8 @@ pub(crate) enum DataStoreUpdateResult { /// How many of a namespace's objects a [`DataStore`] keeps in memory. #[derive(PartialEq, Eq, Debug, Clone, Copy)] pub(crate) enum CacheLimit { + /// Do not keep objects in memory. + Disabled, /// Keep every object in memory. Unbounded, /// Keep at most this many objects in memory. @@ -71,6 +82,18 @@ pub(crate) trait CachePolicy: Send + Sync + 'static { fn cache_limit(&self) -> CacheLimit; } +/// Keeps no objects in memory. +/// +/// Reads always go to the [`KVStore`]. Suitable for namespaces that can grow without bound and do +/// not need a cached working set. +pub(crate) struct KeepNoEntries; + +impl CachePolicy for KeepNoEntries { + fn cache_limit(&self) -> CacheLimit { + CacheLimit::Disabled + } +} + /// Keeps every object of the namespace in memory. /// /// Reads are served entirely from memory and never hit the [`KVStore`]. Required for stores whose @@ -169,6 +192,7 @@ impl LruCache { /// The in-memory part of a [`DataStore`]. enum ObjectCache { + None, KeepAll(HashMap), BoundedLru(LruCache), } @@ -176,6 +200,7 @@ enum ObjectCache { impl ObjectCache { fn new(cache_limit: CacheLimit, objects: Vec) -> Self { match cache_limit { + CacheLimit::Disabled => Self::None, CacheLimit::Unbounded => Self::KeepAll(HashMap::from_iter( objects.into_iter().map(|object| (object.id(), object)), )), @@ -196,6 +221,7 @@ impl ObjectCache { /// Returns the cached object for `id`, marking it as most recently used. fn get(&mut self, id: &SO::Id) -> Option { match self { + Self::None => None, Self::KeepAll(objects) => objects.get(id).cloned(), Self::BoundedLru(lru) => lru.get(id), } @@ -204,6 +230,7 @@ impl ObjectCache { /// Returns the cached object for `id`, without marking it as most recently used. fn peek(&self, id: &SO::Id) -> Option { match self { + Self::None => None, Self::KeepAll(objects) => objects.get(id).cloned(), Self::BoundedLru(lru) => lru.entries.get(id).map(|(object, _)| object.clone()), } @@ -212,6 +239,7 @@ impl ObjectCache { /// Returns whether `id` is cached, without marking it as most recently used. fn contains(&self, id: &SO::Id) -> bool { match self { + Self::None => false, Self::KeepAll(objects) => objects.contains_key(id), Self::BoundedLru(lru) => lru.entries.contains_key(id), } @@ -219,6 +247,7 @@ impl ObjectCache { fn insert(&mut self, id: SO::Id, object: SO) { match self { + Self::None => {}, Self::KeepAll(objects) => { objects.insert(id, object); }, @@ -228,6 +257,7 @@ impl ObjectCache { fn remove(&mut self, id: &SO::Id) { match self { + Self::None => {}, Self::KeepAll(objects) => { objects.remove(id); }, @@ -239,6 +269,7 @@ impl ObjectCache { /// namespace if [`Self::is_keep_all`]. fn filter bool>(&self, f: F) -> Vec { match self { + Self::None => Vec::new(), Self::KeepAll(objects) => objects.values().filter(f).cloned().collect(), Self::BoundedLru(lru) => { lru.entries.values().map(|(object, _)| object).filter(f).cloned().collect() @@ -249,6 +280,7 @@ impl ObjectCache { #[cfg(test)] fn len(&self) -> usize { match self { + Self::None => 0, Self::KeepAll(objects) => objects.len(), Self::BoundedLru(lru) => lru.entries.len(), } @@ -321,33 +353,6 @@ where Ok(()) } - /// Like [`Self::insert`], but when an entry with the object's id already exists, merges the - /// object's full update ([`StorableObject::to_update`]) into it instead of replacing it. - /// - /// Returns whether anything was written. - pub(crate) async fn insert_or_update(&self, object: SO) -> Result { - let _guard = self.mutation_lock.write().await; - - let id = object.id(); - // Note we have to look through to the store here: merging against a cache miss would - // overwrite an evicted object with whatever the caller happens to know about it. - let data_to_persist = match self.lookup(&id).await? { - Some(mut existing_object) => { - existing_object.update(object.to_update()).then_some(existing_object) - }, - None => Some(object), - }; - - match data_to_persist { - Some(updated_object) => { - self.persist(&updated_object).await?; - self.cache.lock().expect("lock").insert(id, updated_object); - Ok(true) - }, - None => Ok(false), - } - } - /// Removes the object stored under `id`, if any. pub(crate) async fn remove(&self, id: &SO::Id) -> Result<(), Error> { let _guard = self.mutation_lock.write().await; @@ -390,23 +395,6 @@ where self.lookup(id).await } - /// Applies `update` to the object stored under its id. - pub(crate) async fn update(&self, update: SO::Update) -> Result { - let _guard = self.mutation_lock.write().await; - - let id = update.id(); - let Some(mut updated_object) = self.lookup(&id).await? else { - return Ok(DataStoreUpdateResult::NotFound); - }; - if !updated_object.update(update) { - return Ok(DataStoreUpdateResult::Unchanged); - } - - self.persist(&updated_object).await?; - self.cache.lock().expect("lock").insert(id, updated_object); - Ok(DataStoreUpdateResult::Updated) - } - /// Atomically transforms the entry for `id` through `f` and persists the result. /// /// `f` receives the current entry (`None` when absent) and returns the new state to write; @@ -463,6 +451,11 @@ where self.contains(id).await } + /// Returns whether this store contains no objects. + pub(crate) async fn is_empty(&self) -> Result { + Ok(self.list_keys_page(None).await?.keys.is_empty()) + } + /// Returns a page of objects, ordered from most recently created to least recently created. /// /// Pass `None` to start at the most recently created object, and the returned @@ -486,29 +479,7 @@ where pub(crate) async fn list_page( &self, page_token: Option, ) -> Result, Error> { - let response = PaginatedKVStore::list_paginated( - &*self.kv_store, - &self.primary_namespace, - &self.secondary_namespace, - page_token, - ) - .await - .map_err(|e| { - log_error!( - self.logger, - "Listing objects under {}/{} failed due to: {}", - &self.primary_namespace, - &self.secondary_namespace, - e - ); - // The backend rejects a token it didn't issue, which is the caller's problem rather - // than a persistence failure. - if e.kind() == ErrorKind::InvalidInput { - Error::InvalidPageToken - } else { - Error::PersistenceFailed - } - })?; + let response = self.list_keys_page(page_token).await?; // Serve whatever we already hold, and note the rest to read below. We take the mutation // lock only for this, so that we observe a consistent view of the cache without holding up @@ -536,6 +507,34 @@ where }) } + async fn list_keys_page( + &self, page_token: Option, + ) -> Result { + PaginatedKVStore::list_paginated( + &*self.kv_store, + &self.primary_namespace, + &self.secondary_namespace, + page_token, + ) + .await + .map_err(|e| { + log_error!( + self.logger, + "Listing objects under {}/{} failed due to: {}", + &self.primary_namespace, + &self.secondary_namespace, + e + ); + // The backend rejects a token it didn't issue, which is the caller's problem rather + // than a persistence failure. + if e.kind() == ErrorKind::InvalidInput { + Error::InvalidPageToken + } else { + Error::PersistenceFailed + } + }) + } + /// Reads the objects we couldn't serve from the cache into their slots in `objects`. /// /// Reads run concurrently but are tracked by slot, as the order in which they finish says @@ -730,6 +729,55 @@ where } } +impl DataStore +where + L::Target: LdkLogger, +{ + /// Like [`Self::insert`], but when an entry with the object's id already exists, merges the + /// object's full update ([`UpdatableObject::to_update`]) into it instead of replacing it. + /// + /// Returns whether anything was written. + pub(crate) async fn insert_or_update(&self, object: SO) -> Result { + let _guard = self.mutation_lock.write().await; + + let id = object.id(); + // Note we have to look through to the store here: merging against a cache miss would + // overwrite an evicted object with whatever the caller happens to know about it. + let data_to_persist = match self.lookup(&id).await? { + Some(mut existing_object) => { + existing_object.update(object.to_update()).then_some(existing_object) + }, + None => Some(object), + }; + + match data_to_persist { + Some(updated_object) => { + self.persist(&updated_object).await?; + self.cache.lock().expect("lock").insert(id, updated_object); + Ok(true) + }, + None => Ok(false), + } + } + + /// Applies `update` to the object stored under its id. + pub(crate) async fn update(&self, update: SO::Update) -> Result { + let _guard = self.mutation_lock.write().await; + + let id = update.id(); + let Some(mut updated_object) = self.lookup(&id).await? else { + return Ok(DataStoreUpdateResult::NotFound); + }; + if !updated_object.update(update) { + return Ok(DataStoreUpdateResult::Unchanged); + } + + self.persist(&updated_object).await?; + self.cache.lock().expect("lock").insert(id, updated_object); + Ok(DataStoreUpdateResult::Updated) + } +} + impl DataStore where L::Target: LdkLogger, @@ -832,11 +880,14 @@ mod tests { impl StorableObject for TestObject { type Id = TestObjectId; - type Update = TestObjectUpdate; fn id(&self) -> Self::Id { self.id } + } + + impl UpdatableObject for TestObject { + type Update = TestObjectUpdate; fn update(&mut self, update: Self::Update) -> bool { let mut updated = false; @@ -1461,6 +1512,51 @@ mod tests { assert_eq!(0, lists.load(Ordering::Relaxed)); } + #[tokio::test] + async fn keep_no_entries_reads_through_without_caching() { + let reads = Arc::new(AtomicUsize::new(0)); + let kv_store: Arc = Arc::new(DynStoreWrapper(CountingStore { + inner: InMemoryStore::new(), + reads: Arc::clone(&reads), + writes: Arc::new(AtomicUsize::new(0)), + removes: Arc::new(AtomicUsize::new(0)), + lists: Arc::new(AtomicUsize::new(0)), + })); + let data_store = new_data_store(kv_store, KeepNoEntries, Vec::new()); + let id = test_id(1); + let object = TestObject::new(id, [23u8; 3]); + + data_store.insert(object).await.unwrap(); + assert_eq!(0, data_store.cached_len()); + + assert_eq!(Some(object), data_store.get(&id).await.unwrap()); + assert_eq!(Some(object), data_store.get(&id).await.unwrap()); + assert!(data_store.contains_key(&id).await.unwrap()); + assert_eq!(3, reads.load(Ordering::Relaxed)); + assert_eq!(0, data_store.cached_len()); + } + + #[tokio::test] + async fn is_empty_lists_keys_without_reading_objects() { + let reads = Arc::new(AtomicUsize::new(0)); + let lists = Arc::new(AtomicUsize::new(0)); + let kv_store: Arc = Arc::new(DynStoreWrapper(CountingStore { + inner: InMemoryStore::new(), + reads: Arc::clone(&reads), + writes: Arc::new(AtomicUsize::new(0)), + removes: Arc::new(AtomicUsize::new(0)), + lists: Arc::clone(&lists), + })); + let data_store = new_data_store(kv_store, KeepNoEntries, Vec::new()); + + assert!(data_store.is_empty().await.unwrap()); + data_store.insert(TestObject::new(test_id(1), [23u8; 3])).await.unwrap(); + assert!(!data_store.is_empty().await.unwrap()); + + assert_eq!(0, reads.load(Ordering::Relaxed)); + assert_eq!(2, lists.load(Ordering::Relaxed)); + } + #[tokio::test] async fn lru_evicts_least_recently_used() { let (data_store, _kv_store, ids) = new_lru_store_with_objects(2, 3).await; diff --git a/src/error.rs b/src/error.rs index 485f944c2..d187c6820 100644 --- a/src/error.rs +++ b/src/error.rs @@ -89,6 +89,10 @@ pub enum Error { InvalidNodeId, /// The given payment id is invalid. InvalidPaymentId, + /// The given forwarded payment id is invalid. + InvalidForwardedPaymentId, + /// The given channel-pair forwarding statistics id is invalid. + InvalidChannelPairForwardingStatsId, /// The given payment hash is invalid. InvalidPaymentHash, /// The given payment pre-image is invalid. @@ -194,6 +198,12 @@ impl fmt::Display for Error { Self::InvalidOfferId => write!(f, "The given offer id is invalid."), Self::InvalidNodeId => write!(f, "The given node id is invalid."), Self::InvalidPaymentId => write!(f, "The given payment id is invalid."), + Self::InvalidForwardedPaymentId => { + write!(f, "The given forwarded payment id is invalid.") + }, + Self::InvalidChannelPairForwardingStatsId => { + write!(f, "The given channel-pair forwarding statistics id is invalid.") + }, Self::InvalidPaymentHash => write!(f, "The given payment hash is invalid."), Self::InvalidPaymentPreimage => write!(f, "The given payment preimage is invalid."), Self::InvalidPaymentSecret => write!(f, "The given payment secret is invalid."), diff --git a/src/event.rs b/src/event.rs index 0a3569755..4b689fbde 100644 --- a/src/event.rs +++ b/src/event.rs @@ -48,6 +48,7 @@ use crate::liquidity::LiquiditySource; 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::forwarding_store::{ForwardRecord, ForwardingStore}; use crate::payment::store::{ PaymentDetails, PaymentDetailsUpdate, PaymentDirection, PaymentKind, PaymentStatus, }; @@ -551,6 +552,7 @@ where network_graph: Arc, liquidity_source: Arc>>, payment_store: Arc, + forwarding_store: Arc, peer_store: Arc>, keys_manager: Arc, static_invoice_store: Option, @@ -572,10 +574,10 @@ where channel_manager: Arc, connection_manager: Arc>, output_sweeper: Arc, network_graph: Arc, liquidity_source: Arc>>, payment_store: Arc, - peer_store: Arc>, keys_manager: Arc, - static_invoice_store: Option, onion_messenger: Arc, - om_mailbox: Option>, prober: Option>, - runtime: Arc, logger: L, config: Arc, + forwarding_store: Arc, peer_store: Arc>, + keys_manager: Arc, static_invoice_store: Option, + onion_messenger: Arc, om_mailbox: Option>, + prober: Option>, runtime: Arc, logger: L, config: Arc, ) -> Self { Self { event_queue, @@ -587,6 +589,7 @@ where network_graph, liquidity_source, payment_store, + forwarding_store, peer_store, keys_manager, static_invoice_store, @@ -1751,6 +1754,18 @@ where .await; } + self.forwarding_store + .record_forward(ForwardRecord { + prev_htlcs: &prev_htlcs, + next_htlcs: &next_htlcs, + total_fee_earned_msat, + skimmed_fee_msat, + claim_from_onchain_tx, + outbound_amount_forwarded_msat, + }) + .await + .map_err(|_| ReplayEvent())?; + let event = Event::PaymentForwarded { prev_htlcs: prev_htlcs.into_iter().map(HTLCLocator::from).collect(), next_htlcs: next_htlcs.into_iter().map(HTLCLocator::from).collect(), @@ -2237,6 +2252,16 @@ mod tests { use crate::payment::store::LSPS2Parameters; use crate::types::DynStoreWrapper; + fn ldk_htlc_locator(channel_byte: u8) -> LdkHtlcLocator { + LdkHtlcLocator { + channel_id: ChannelId([channel_byte; 32]), + htlc_id: None, + amount_msat: Some(channel_byte as u64), + user_channel_id: Some(channel_byte as u128), + node_id: None, + } + } + #[test] fn lsps2_payment_metadata_decodes_total_fee_limit() { let metadata = PaymentMetadata { @@ -2461,6 +2486,52 @@ mod tests { ); } + #[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)), HTLCLocator::from(ldk_htlc_locator(2))]; + let next_htlcs = vec![HTLCLocator::from(ldk_htlc_locator(3))]; + 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: 800, + }) + ); + } + + #[test] + fn payment_forwarded_event_roundtrips() { + let event = Event::PaymentForwarded { + prev_htlcs: vec![HTLCLocator::from(ldk_htlc_locator(1))], + next_htlcs: vec![HTLCLocator::from(ldk_htlc_locator(2))], + total_fee_earned_msat: Some(200), + skimmed_fee_msat: None, + claim_from_onchain_tx: false, + outbound_amount_forwarded_msat: 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 d213f9d51..a153a77ef 100644 --- a/src/ffi/types.rs +++ b/src/ffi/types.rs @@ -159,6 +159,7 @@ pub use crate::config::default_config; use crate::error::Error; pub use crate::liquidity::LSPS1OrderStatus; pub use crate::logger::{LogLevel, LogRecord, LogWriter}; +use crate::payment::{ChannelPairForwardingStatsId, ForwardedPaymentId}; pub use crate::probing::ProbingConfig; use crate::{hex_utils, SocketAddress, UserChannelId}; @@ -1060,6 +1061,26 @@ uniffi::custom_type!(PaymentId, String, { }, }); +uniffi::custom_type!(ForwardedPaymentId, String, { + remote, + try_lift: |val| { + Ok(ForwardedPaymentId::from_str(&val)?) + }, + lower: |obj| { + obj.to_string() + }, +}); + +uniffi::custom_type!(ChannelPairForwardingStatsId, String, { + remote, + try_lift: |val| { + Ok(ChannelPairForwardingStatsId::from_str(&val)?) + }, + lower: |obj| { + obj.to_string() + }, +}); + uniffi::custom_type!(PaymentHash, String, { remote, try_lift: |val| { diff --git a/src/io/mod.rs b/src/io/mod.rs index c11475c43..b7e4d2131 100644 --- a/src/io/mod.rs +++ b/src/io/mod.rs @@ -37,6 +37,15 @@ 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 = ""; +/// Forwarded payment information is persisted under this primary namespace. +pub(crate) const FORWARDED_PAYMENT_PERSISTENCE_PRIMARY_NAMESPACE: &str = "forwarded_payments"; +pub(crate) const FORWARDED_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE: &str = "details"; +pub(crate) const FORWARDED_PAYMENT_REPLAY_MARKER_PERSISTENCE_SECONDARY_NAMESPACE: &str = + "replay_markers"; +pub(crate) const CHANNEL_FORWARDING_STATS_PERSISTENCE_SECONDARY_NAMESPACE: &str = "channel_stats"; +pub(crate) const CHANNEL_PAIR_FORWARDING_STATS_PERSISTENCE_SECONDARY_NAMESPACE: &str = + "channel_pair_stats"; + /// 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/lib.rs b/src/lib.rs index b6f790767..59fd362e2 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -175,9 +175,11 @@ 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; +pub use payment::forwarding_store::aggregate_channel_pair_stats; +use payment::forwarding_store::{run_forwarded_payment_aggregation, ForwardingStore}; use payment::{ - Bolt11Payment, Bolt12Payment, OnchainPayment, PaymentDetails, PaymentDetailsPage, - SpontaneousPayment, + Bolt11Payment, Bolt12Payment, ForwardingAnalytics, OnchainPayment, PaymentDetails, + PaymentDetailsPage, SpontaneousPayment, }; #[cfg(feature = "unified-payments")] use payment::{HRNResolver, UnifiedPayment}; @@ -271,6 +273,8 @@ pub struct Node { scorer: Arc>, peer_store: Arc>>, payment_store: Arc, + forwarding_store: Arc, + forwarded_payment_aggregation_retention_secs: u64, lnurl_auth: Arc, is_running: Arc>, node_metrics: Arc, @@ -654,6 +658,14 @@ impl Node { chain_source.continuously_process_broadcast_queue(stop_tx_bcast).await }); + let retention_secs = self.forwarded_payment_aggregation_retention_secs; + let stop_aggregation = self.stop_sender.subscribe(); + let forwarding_store = Arc::clone(&self.forwarding_store); + self.runtime.spawn_cancellable_background_task(async move { + run_forwarded_payment_aggregation(stop_aggregation, forwarding_store, retention_secs) + .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))), @@ -678,6 +690,7 @@ impl Node { Arc::clone(&self.network_graph), Arc::clone(&self.liquidity_source), Arc::clone(&self.payment_store), + Arc::clone(&self.forwarding_store), Arc::clone(&self.peer_store), Arc::clone(&self.keys_manager), static_invoice_store, @@ -1200,6 +1213,26 @@ impl Node { } impl Node { + /// Returns a handler allowing to query forwarded payments and forwarding statistics. + #[cfg(not(feature = "uniffi"))] + pub fn forwarding_analytics(&self) -> ForwardingAnalytics { + ForwardingAnalytics::new( + Arc::clone(&self.runtime), + Arc::clone(&self.forwarding_store), + Arc::clone(&self.config), + ) + } + + /// Returns a handler allowing to query forwarded payments and forwarding statistics. + #[cfg(feature = "uniffi")] + pub fn forwarding_analytics(&self) -> Arc { + Arc::new(ForwardingAnalytics::new( + Arc::clone(&self.runtime), + Arc::clone(&self.forwarding_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 diff --git a/src/payment/forwarding.rs b/src/payment/forwarding.rs new file mode 100644 index 000000000..2a50f24c8 --- /dev/null +++ b/src/payment/forwarding.rs @@ -0,0 +1,388 @@ +// 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::fmt; +use std::str::FromStr; +use std::sync::Arc; + +use bitcoin::secp256k1::PublicKey; +use lightning::impl_writeable_tlv_based; +use lightning::ln::types::ChannelId; +use lightning::util::ser::{Readable, Writeable, Writer}; + +use crate::config::{Config, ForwardedPaymentTrackingMode}; +use crate::error::Error; +use crate::ffi::{maybe_deref, maybe_wrap}; +use crate::hex_utils; +use crate::runtime::Runtime; +use crate::UserChannelId; + +use super::forwarding_store::ForwardingStore; +use super::PageToken; + +fn parse_forwarding_id(id: &str) -> Option<[u8; 32]> { + hex_utils::to_vec(id)?.try_into().ok() +} + +/// An identifier for a forwarded payment. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] +pub struct ForwardedPaymentId(pub [u8; 32]); + +impl FromStr for ForwardedPaymentId { + type Err = Error; + + fn from_str(id: &str) -> Result { + parse_forwarding_id(id).map(Self).ok_or(Error::InvalidForwardedPaymentId) + } +} + +impl fmt::Display for ForwardedPaymentId { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(&hex_utils::to_string(&self.0)) + } +} + +impl Writeable for ForwardedPaymentId { + fn write(&self, writer: &mut W) -> Result<(), lightning::io::Error> { + Ok(self.0.write(writer)?) + } +} + +impl Readable for ForwardedPaymentId { + fn read( + reader: &mut R, + ) -> Result { + Ok(Self(Readable::read(reader)?)) + } +} + +/// An identifier for an aggregated channel-pair forwarding statistics bucket. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] +pub struct ChannelPairForwardingStatsId(pub [u8; 32]); + +impl FromStr for ChannelPairForwardingStatsId { + type Err = Error; + + fn from_str(id: &str) -> Result { + parse_forwarding_id(id).map(Self).ok_or(Error::InvalidChannelPairForwardingStatsId) + } +} + +impl fmt::Display for ChannelPairForwardingStatsId { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(&hex_utils::to_string(&self.0)) + } +} + +impl Writeable for ChannelPairForwardingStatsId { + fn write(&self, writer: &mut W) -> Result<(), lightning::io::Error> { + Ok(self.0.write(writer)?) + } +} + +impl Readable for ChannelPairForwardingStatsId { + fn read( + reader: &mut R, + ) -> Result { + Ok(Self(Readable::read(reader)?)) + } +} + +/// Details of a payment that has been forwarded through this node. +/// +/// A forward is recorded only when it consisted of exactly one incoming and one outgoing HTLC and +/// LDK reported a total fee. LDK reports no fee when the incoming channel was force-closed and the +/// funds are claimed on chain, because the on-chain fees are not yet known at that point, so those +/// forwards are not recorded at all. +#[derive(Clone, Debug, PartialEq, Eq)] +#[cfg_attr(feature = "uniffi", derive(uniffi::Record))] +pub struct ForwardedPaymentDetails { + /// An opaque 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. + /// + /// This is the share of [`Self::total_fee_earned_msat`] that was withheld in addition to the + /// forwarding fee, not an amount earned on top of it. Adding the two would double-count. + 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), +}); + +/// 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. + /// + /// A single record covers this channel in both roles. A forward contributes its fee here when + /// this was the incoming channel, and contributes nothing when this was the outgoing channel, + /// because fees are attributed to the incoming side. + pub total_fee_earned_msat: Option, + /// Total skimmed fees earned through this channel, in millisatoshis. + /// + /// This is the share of [`Self::total_fee_earned_msat`] that was withheld in addition to the + /// forwarding fee, not an amount earned on top of it. Adding the two would double-count. + pub total_skimmed_fee_msat: u64, + /// Number of forwarded HTLCs that the next hop claimed from an on-chain transaction. + /// + /// A forward contributes here when this was the outgoing channel. + 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), +}); + +/// Aggregated statistics for a specific channel pair. +#[derive(Clone, Debug, PartialEq, Eq)] +#[cfg_attr(feature = "uniffi", derive(uniffi::Record))] +pub struct ChannelPairForwardingStats { + /// An opaque identifier for this channel-pair bucket. + pub id: ChannelPairForwardingStatsId, + /// 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. + /// + /// This is the share of [`Self::total_fee_earned_msat`] that was withheld in addition to the + /// forwarding fee, not an amount earned on top of it. Adding the two would double-count. + pub total_skimmed_fee_msat: u64, + /// Number of forwarded HTLCs that the next hop claimed from an on-chain transaction. + 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), +}); + +/// 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_analytics`]. +/// +/// [`Node::forwarding_analytics`]: crate::Node::forwarding_analytics +#[cfg_attr(feature = "uniffi", derive(uniffi::Object))] +pub struct ForwardingAnalytics { + runtime: Arc, + forwarding_store: Arc, + config: Arc, +} + +impl ForwardingAnalytics { + pub(crate) fn new( + runtime: Arc, forwarding_store: Arc, config: Arc, + ) -> Self { + Self { runtime, forwarding_store, config } + } + + /// Retrieves all channel forwarding statistics that match the given predicate. + pub fn list_channel_stats_with_filter bool>( + &self, f: F, + ) -> Vec { + self.runtime.block_on(self.forwarding_store.list_channel_stats_with_filter(f)) + } +} + +#[cfg_attr(feature = "uniffi", uniffi::export)] +impl ForwardingAnalytics { + /// 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 using its opaque identifier. + /// + /// The identifier is returned in [`ForwardedPaymentDetails::id`]. + pub fn payment( + &self, forwarded_payment_id: &ForwardedPaymentId, + ) -> Result, Error> { + self.runtime.block_on(self.forwarding_store.payment(forwarded_payment_id)) + } + + /// Retrieves a page of forwarded payments from the underlying paginated store. + pub fn list_payments( + &self, page_token: Option, + ) -> Result { + let ldk_page_token = page_token.as_ref().map(|token| maybe_deref(token).clone()); + let page = self.runtime.block_on(self.forwarding_store.list_payments(ldk_page_token))?; + Ok(ForwardedPaymentDetailsPage { + payments: page.objects, + next_page_token: page.next_page_token.map(maybe_wrap), + }) + } + + /// Retrieve the forwarding statistics for a specific channel. + pub fn channel_stats( + &self, channel_id: &ChannelId, + ) -> Result, Error> { + self.runtime.block_on(self.forwarding_store.channel_stats(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 ldk_page_token = page_token.as_ref().map(|token| maybe_deref(token).clone()); + let page = + self.runtime.block_on(self.forwarding_store.list_channel_stats(ldk_page_token))?; + Ok(ChannelForwardingStatsPage { + stats: page.objects, + next_page_token: page.next_page_token.map(maybe_wrap), + }) + } + + /// 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 ldk_page_token = page_token.as_ref().map(|token| maybe_deref(token).clone()); + let page = + self.runtime.block_on(self.forwarding_store.list_channel_pair_stats(ldk_page_token))?; + Ok(ChannelPairForwardingStatsPage { + stats: page.objects, + next_page_token: page.next_page_token.map(maybe_wrap), + }) + } +} diff --git a/src/payment/forwarding_store.rs b/src/payment/forwarding_store.rs new file mode 100644 index 000000000..722a9138d --- /dev/null +++ b/src/payment/forwarding_store.rs @@ -0,0 +1,1758 @@ +// 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. + +use std::collections::{HashMap, HashSet}; +use std::sync::Arc; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use bitcoin::hashes::{sha256, Hash}; +use bitcoin::secp256k1::PublicKey; +use lightning::events::HTLCLocator; +use lightning::impl_writeable_tlv_based; +use lightning::ln::types::ChannelId; +use lightning::util::logger::Logger as _; +use lightning::util::persist::PageToken; + +use super::forwarding::{ + ChannelForwardingStats, ChannelPairForwardingStats, ChannelPairForwardingStatsId, + ForwardedPaymentDetails, ForwardedPaymentId, +}; +use crate::config::ForwardedPaymentTrackingMode; +use crate::data_store::{ + DataStore, DataStorePage, KeepAllEntries, KeepNoEntries, StorableObject, StorableObjectId, + StorableObjectUpdate, UpdatableObject, +}; +use crate::hex_utils; +use crate::io::{ + CHANNEL_FORWARDING_STATS_PERSISTENCE_SECONDARY_NAMESPACE, + CHANNEL_PAIR_FORWARDING_STATS_PERSISTENCE_SECONDARY_NAMESPACE, + FORWARDED_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, + FORWARDED_PAYMENT_PERSISTENCE_PRIMARY_NAMESPACE, + FORWARDED_PAYMENT_REPLAY_MARKER_PERSISTENCE_SECONDARY_NAMESPACE, +}; +use crate::logger::{log_debug, log_error, Logger}; +use crate::types::{ + ChannelForwardingStatsStore, ChannelPairForwardingStatsStore, DynStore, ForwardedPaymentStore, +}; +use crate::{Error, UserChannelId}; + +pub(crate) const FORWARDED_PAYMENT_AGGREGATION_BUCKET_SIZE_SECS: u64 = 60 * 60; + +impl StorableObjectId for ForwardedPaymentId { + fn encode_to_hex_str(&self) -> String { + hex_utils::to_string(&self.0) + } + + fn decode_from_hex_str(s: &str) -> Option { + s.parse().ok() + } +} + +impl StorableObjectId for ChannelPairForwardingStatsId { + fn encode_to_hex_str(&self) -> String { + hex_utils::to_string(&self.0) + } + + fn decode_from_hex_str(s: &str) -> Option { + s.parse().ok() + } +} + +pub(crate) struct ForwardRecord<'a> { + pub prev_htlcs: &'a [HTLCLocator], + pub next_htlcs: &'a [HTLCLocator], + pub total_fee_earned_msat: Option, + pub skimmed_fee_msat: Option, + pub claim_from_onchain_tx: bool, + pub outbound_amount_forwarded_msat: u64, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +struct ForwardedPaymentReplayMarker { + id: ForwardedPaymentId, +} + +impl_writeable_tlv_based!(ForwardedPaymentReplayMarker, { (0, id, required) }); + +impl StorableObject for ForwardedPaymentReplayMarker { + type Id = ForwardedPaymentId; + + fn id(&self) -> Self::Id { + self.id + } +} + +type ForwardedPaymentReplayMarkerStore = + DataStore, KeepNoEntries>; + +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) struct StoredChannelForwardingStats { + channel_id: ChannelId, + counterparty_node_id: Option, + inbound_payments_forwarded: u64, + outbound_payments_forwarded: u64, + total_inbound_amount_msat: u64, + total_outbound_amount_msat: u64, + total_fee_earned_msat: Option, + total_skimmed_fee_msat: u64, + onchain_claims_count: u64, + first_forwarded_at_timestamp: u64, + last_forwarded_at_timestamp: u64, + last_inbound_forward_id: Option, + last_outbound_forward_id: Option, +} + +impl_writeable_tlv_based!(StoredChannelForwardingStats, { + (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), + (22, last_inbound_forward_id, option), + (24, last_outbound_forward_id, option), +}); + +impl From for ChannelForwardingStats { + fn from(stats: StoredChannelForwardingStats) -> Self { + Self { + channel_id: stats.channel_id, + counterparty_node_id: stats.counterparty_node_id, + inbound_payments_forwarded: stats.inbound_payments_forwarded, + outbound_payments_forwarded: stats.outbound_payments_forwarded, + total_inbound_amount_msat: stats.total_inbound_amount_msat, + total_outbound_amount_msat: stats.total_outbound_amount_msat, + total_fee_earned_msat: stats.total_fee_earned_msat, + total_skimmed_fee_msat: stats.total_skimmed_fee_msat, + onchain_claims_count: stats.onchain_claims_count, + first_forwarded_at_timestamp: stats.first_forwarded_at_timestamp, + last_forwarded_at_timestamp: stats.last_forwarded_at_timestamp, + } + } +} + +impl From for StoredChannelForwardingStats { + fn from(stats: ChannelForwardingStats) -> Self { + Self { + channel_id: stats.channel_id, + counterparty_node_id: stats.counterparty_node_id, + inbound_payments_forwarded: stats.inbound_payments_forwarded, + outbound_payments_forwarded: stats.outbound_payments_forwarded, + total_inbound_amount_msat: stats.total_inbound_amount_msat, + total_outbound_amount_msat: stats.total_outbound_amount_msat, + total_fee_earned_msat: stats.total_fee_earned_msat, + total_skimmed_fee_msat: stats.total_skimmed_fee_msat, + onchain_claims_count: stats.onchain_claims_count, + first_forwarded_at_timestamp: stats.first_forwarded_at_timestamp, + last_forwarded_at_timestamp: stats.last_forwarded_at_timestamp, + last_inbound_forward_id: None, + last_outbound_forward_id: None, + } + } +} + +/// Owns all forwarding data stores and the operations that must keep them consistent. +pub(crate) struct ForwardingStore { + details: ForwardedPaymentStore, + replay_markers: ForwardedPaymentReplayMarkerStore, + channel_stats: ChannelForwardingStatsStore, + channel_pair_stats: ChannelPairForwardingStatsStore, + tracking_mode: ForwardedPaymentTrackingMode, + logger: Arc, +} + +impl ForwardingStore { + pub(crate) fn new( + channel_stats: Vec, + tracking_mode: ForwardedPaymentTrackingMode, kv_store: Arc, logger: Arc, + ) -> Self { + Self { + details: ForwardedPaymentStore::new( + Vec::new(), + KeepNoEntries, + FORWARDED_PAYMENT_PERSISTENCE_PRIMARY_NAMESPACE.to_string(), + FORWARDED_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE.to_string(), + Arc::clone(&kv_store), + Arc::clone(&logger), + ), + replay_markers: DataStore::new( + Vec::new(), + KeepNoEntries, + FORWARDED_PAYMENT_PERSISTENCE_PRIMARY_NAMESPACE.to_string(), + FORWARDED_PAYMENT_REPLAY_MARKER_PERSISTENCE_SECONDARY_NAMESPACE.to_string(), + Arc::clone(&kv_store), + Arc::clone(&logger), + ), + channel_stats: ChannelForwardingStatsStore::new( + channel_stats, + KeepAllEntries, + FORWARDED_PAYMENT_PERSISTENCE_PRIMARY_NAMESPACE.to_string(), + CHANNEL_FORWARDING_STATS_PERSISTENCE_SECONDARY_NAMESPACE.to_string(), + Arc::clone(&kv_store), + Arc::clone(&logger), + ), + channel_pair_stats: ChannelPairForwardingStatsStore::new( + Vec::new(), + KeepNoEntries, + FORWARDED_PAYMENT_PERSISTENCE_PRIMARY_NAMESPACE.to_string(), + CHANNEL_PAIR_FORWARDING_STATS_PERSISTENCE_SECONDARY_NAMESPACE.to_string(), + kv_store, + Arc::clone(&logger), + ), + tracking_mode, + logger, + } + } + + pub(crate) async fn record_forward(&self, forward: ForwardRecord<'_>) -> Result<(), Error> { + let ([prev_htlc], [next_htlc], Some(fee_earned_msat), Some(prev_htlc_id)) = ( + forward.prev_htlcs, + forward.next_htlcs, + forward.total_fee_earned_msat, + forward.prev_htlcs.first().and_then(|htlc| htlc.htlc_id), + ) else { + log_debug!( + self.logger, + "Skipping forwarding payment tracking for forward with {} inbound and {} outbound HTLC(s), because tracking requires a single HTLC in each direction, a known fee, and an incoming HTLC ID", + forward.prev_htlcs.len(), + forward.next_htlcs.len() + ); + return Ok(()); + }; + + let forward_id = forwarded_payment_id(&prev_htlc.channel_id, prev_htlc_id); + if self.replay_markers.contains_key(&forward_id).await.map_err(|e| { + log_error!(self.logger, "Failed to check forwarded payment replay marker: {e}"); + e + })? { + return Ok(()); + } + let details_id = matches!(self.tracking_mode, ForwardedPaymentTrackingMode::Detailed) + .then_some(forward_id); + let forwarded_at_timestamp = SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("current time should not be earlier than the Unix epoch") + .as_secs(); + let inbound_amount_msat = + forward.outbound_amount_forwarded_msat.saturating_add(fee_earned_msat); + + // Store the stable forward ID with each directional update. If a later write fails, a replay + // can finish the remaining updates without applying a successful update twice. + let mut outbound_stats = StoredChannelForwardingStats::from(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: forward.outbound_amount_forwarded_msat, + total_fee_earned_msat: Some(0), + total_skimmed_fee_msat: 0, + onchain_claims_count: u64::from(forward.claim_from_onchain_tx), + first_forwarded_at_timestamp: forwarded_at_timestamp, + last_forwarded_at_timestamp: forwarded_at_timestamp, + }); + outbound_stats.last_outbound_forward_id = Some(forward_id); + self.channel_stats.insert_or_update(outbound_stats).await.map_err(|e| { + log_error!(self.logger, "Failed to update outbound channel forwarding stats: {e}"); + e + })?; + + let mut inbound_stats = StoredChannelForwardingStats::from(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: inbound_amount_msat, + total_outbound_amount_msat: 0, + total_fee_earned_msat: Some(fee_earned_msat), + total_skimmed_fee_msat: forward.skimmed_fee_msat.unwrap_or(0), + onchain_claims_count: 0, + first_forwarded_at_timestamp: forwarded_at_timestamp, + last_forwarded_at_timestamp: forwarded_at_timestamp, + }); + inbound_stats.last_inbound_forward_id = Some(forward_id); + self.channel_stats.insert_or_update(inbound_stats).await.map_err(|e| { + log_error!(self.logger, "Failed to update inbound channel forwarding stats: {e}"); + e + })?; + + if let Some(id) = details_id { + self.details + .insert(ForwardedPaymentDetails { + id, + 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(inbound_amount_msat), + total_fee_earned_msat: Some(fee_earned_msat), + skimmed_fee_msat: forward.skimmed_fee_msat, + claim_from_onchain_tx: forward.claim_from_onchain_tx, + outbound_amount_forwarded_msat: Some(forward.outbound_amount_forwarded_msat), + forwarded_at_timestamp, + }) + .await + .map_err(|e| { + log_error!(self.logger, "Failed to store forwarded payment: {e}"); + e + })?; + } + + // Keep this marker after the event is handled. LDK can replay an older event after later + // events have replaced the directional retry tokens, and it provides no callback after its + // handled-event state is durable. + self.replay_markers.insert(ForwardedPaymentReplayMarker { id: forward_id }).await.map_err( + |e| { + log_error!(self.logger, "Failed to store forwarded payment replay marker: {e}"); + e + }, + )?; + + Ok(()) + } + + pub(crate) async fn payment( + &self, id: &ForwardedPaymentId, + ) -> Result, Error> { + self.details.get(id).await + } + + pub(crate) async fn list_payments( + &self, page_token: Option, + ) -> Result, Error> { + self.details.list_page(page_token).await + } + + pub(crate) async fn channel_stats( + &self, id: &ChannelId, + ) -> Result, Error> { + Ok(self.channel_stats.get(id).await?.map(Into::into)) + } + + pub(crate) async fn list_channel_stats( + &self, page_token: Option, + ) -> Result, Error> { + let page = self.channel_stats.list_page(page_token).await?; + Ok(DataStorePage { + objects: page.objects.into_iter().map(Into::into).collect(), + next_page_token: page.next_page_token, + }) + } + + pub(crate) async fn list_channel_stats_with_filter< + F: FnMut(&&ChannelForwardingStats) -> bool, + >( + &self, f: F, + ) -> Vec { + let stats: Vec<_> = + self.channel_stats.list_filter(|_| true).await.into_iter().map(Into::into).collect(); + let mut f = f; + stats.into_iter().filter(|stats| f(&stats)).collect() + } + + pub(crate) async fn list_channel_pair_stats( + &self, page_token: Option, + ) -> Result, Error> { + self.channel_pair_stats.list_page(page_token).await + } + + pub(crate) async fn aggregate_expired(&self, retention_secs: u64) -> Result<(u64, u64), Error> { + aggregate_expired_forwarded_payments( + &self.details, + &self.replay_markers, + &self.channel_pair_stats, + retention_secs, + &self.logger, + ) + .await + } +} + +fn forwarded_payment_id(channel_id: &ChannelId, htlc_id: u64) -> ForwardedPaymentId { + let mut bytes = [0u8; 40]; + bytes[..32].copy_from_slice(&channel_id.0); + bytes[32..].copy_from_slice(&htlc_id.to_be_bytes()); + ForwardedPaymentId(sha256::Hash::hash(&bytes).to_byte_array()) +} + +impl StorableObject for ForwardedPaymentDetails { + type Id = ForwardedPaymentId; + + fn id(&self) -> Self::Id { + self.id + } +} + +fn channel_pair_stats_id( + prev: &ChannelId, next: &ChannelId, bucket_start_timestamp: u64, +) -> ChannelPairForwardingStatsId { + // Hash the ordered channel pair and bucket start. The bucket size is fixed and therefore does + // not need to be part of the ID. Keeping the channels ordered preserves forwarding direction, + // unlike combining them with XOR. + let mut bytes = [0u8; 72]; + bytes[0..32].copy_from_slice(&prev.0); + bytes[32..64].copy_from_slice(&next.0); + bytes[64..72].copy_from_slice(&bucket_start_timestamp.to_be_bytes()); + ChannelPairForwardingStatsId(sha256::Hash::hash(&bytes).to_byte_array()) +} + +#[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, + pub last_inbound_forward_id: Option, + pub last_outbound_forward_id: Option, +} + +impl StorableObjectUpdate for ChannelForwardingStatsUpdate { + fn id(&self) -> ChannelId { + self.channel_id + } +} + +impl StorableObjectId for ChannelId { + fn encode_to_hex_str(&self) -> String { + hex_utils::to_string(&self.0) + } + + fn decode_from_hex_str(s: &str) -> Option { + let bytes = hex_utils::to_vec(s)?; + Some(ChannelId(bytes.try_into().ok()?)) + } +} + +impl StorableObject for StoredChannelForwardingStats { + type Id = ChannelId; + + fn id(&self) -> Self::Id { + self.channel_id + } +} + +impl UpdatableObject for StoredChannelForwardingStats { + type Update = ChannelForwardingStatsUpdate; + + 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; + } + let apply_inbound = update.inbound_payments_increment > 0 + && (update.last_inbound_forward_id.is_none() + || update.last_inbound_forward_id != self.last_inbound_forward_id); + let apply_outbound = update.outbound_payments_increment > 0 + && (update.last_outbound_forward_id.is_none() + || update.last_outbound_forward_id != self.last_outbound_forward_id); + if apply_inbound { + self.inbound_payments_forwarded += update.inbound_payments_increment; + self.total_inbound_amount_msat += update.inbound_amount_increment_msat; + match (self.total_fee_earned_msat.as_mut(), update.fee_earned_increment_msat) { + (Some(total), Some(increment)) => *total += increment, + (Some(_), None) => self.total_fee_earned_msat = None, + _ => {}, + } + self.total_skimmed_fee_msat += update.skimmed_fee_increment_msat; + self.last_inbound_forward_id = update.last_inbound_forward_id; + updated = true; + } + if apply_outbound { + self.outbound_payments_forwarded += update.outbound_payments_increment; + self.total_outbound_amount_msat += update.outbound_amount_increment_msat; + self.onchain_claims_count += update.onchain_claims_increment; + self.last_outbound_forward_id = update.last_outbound_forward_id; + 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, + last_inbound_forward_id: self.last_inbound_forward_id, + last_outbound_forward_id: self.last_outbound_forward_id, + } + } +} + +impl StorableObject for ChannelPairForwardingStats { + type Id = ChannelPairForwardingStatsId; + + fn id(&self) -> Self::Id { + self.id + } +} + +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) +} + +async fn aggregate_forwarded_payments_and_log( + forwarding_store: &ForwardingStore, retention_secs: u64, +) { + match forwarding_store.aggregate_expired(retention_secs).await { + Ok((pair_count, payment_count)) if pair_count > 0 => { + log_debug!( + forwarding_store.logger, + "Aggregated {} forwarded payments into {} channel pair buckets", + payment_count, + pair_count + ); + }, + Ok((0, payment_count)) if payment_count > 0 => { + log_debug!( + forwarding_store.logger, + "Removed {} forwarded payment details from previously aggregated buckets", + payment_count + ); + }, + Err(e) => { + log_error!(forwarding_store.logger, "Forwarded payment aggregation failed: {}", e) + }, + _ => {}, + } +} + +pub(crate) async fn run_forwarded_payment_aggregation( + mut stop_receiver: tokio::sync::watch::Receiver<()>, forwarding_store: Arc, + retention_secs: u64, +) { + if retention_secs == 0 { + match forwarding_store.details.is_empty().await { + Ok(true) => return, + Ok(false) => {}, + Err(e) => log_error!( + forwarding_store.logger, + "Failed to check forwarded payment store: {}", + e + ), + } + } + + aggregate_forwarded_payments_and_log(&forwarding_store, retention_secs).await; + + if retention_secs == 0 { + match forwarding_store.details.is_empty().await { + Ok(true) => return, + Ok(false) => {}, + Err(e) => log_error!( + forwarding_store.logger, + "Failed to check forwarded payment store: {}", + e + ), + } + } + + let period = Duration::from_secs(FORWARDED_PAYMENT_AGGREGATION_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, + FORWARDED_PAYMENT_AGGREGATION_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() => { + aggregate_forwarded_payments_and_log(&forwarding_store, retention_secs).await; + if retention_secs == 0 { + match forwarding_store.details.is_empty().await { + Ok(true) => break, + Ok(false) => {}, + Err(e) => log_error!(forwarding_store.logger, "Failed to check forwarded payment store: {}", e), + } + } + } + } + } +} + +/// Aggregate forwarded payments older than the configured retention period into fixed-width +/// channel-pair statistics buckets. +async fn aggregate_expired_forwarded_payments( + forwarded_payment_store: &ForwardedPaymentStore, + replay_marker_store: &ForwardedPaymentReplayMarkerStore, + channel_pair_stats_store: &ChannelPairForwardingStatsStore, retention_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, + replay_marker_store, + channel_pair_stats_store, + FORWARDED_PAYMENT_AGGREGATION_BUCKET_SIZE_SECS, + retention_secs, + now, + logger, + ) + .await +} + +async fn aggregate_expired_forwarded_payments_at( + forwarded_payment_store: &ForwardedPaymentStore, + replay_marker_store: &ForwardedPaymentReplayMarkerStore, + channel_pair_stats_store: &ChannelPairForwardingStatsStore, bucket_size_secs: u64, + retention_secs: u64, now: u64, logger: &Arc, +) -> Result<(u64, u64), Error> { + if bucket_size_secs == 0 { + return Ok((0, 0)); + } + let retention_cutoff = now.saturating_sub(retention_secs); + let oldest_retained_bucket_start = + (retention_cutoff / bucket_size_secs).saturating_mul(bucket_size_secs); + + // Retain details for at least the configured period. We only aggregate complete buckets, so + // details may remain for up to one additional bucket width. 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. + // + // The scan below runs unlocked, so that a pass over a large store never stalls event handling. + // What keeps a record out of a bucket we have already closed is that width of slack, not any + // locking: a record would have to be stamped and then take longer than a whole bucket to be + // written. Event handling is serialized, so long before that the node has stopped forwarding + // anything at all. Insertions interleaving between pages are therefore fine, and a straggler we + // miss is found by the next pass, whose bucket marker exists by then, so it is deleted rather + // than counted twice. + let mut bucket_groups: HashMap<(ChannelId, ChannelId, u64), Vec> = + HashMap::new(); + let mut incomplete_buckets = HashSet::new(); + let mut page_token = None; + loop { + let page = forwarded_payment_store.list_page(page_token).await?; + for payment in page.objects { + if payment.forwarded_at_timestamp >= oldest_retained_bucket_start { + continue; + } + let bucket_start = + (payment.forwarded_at_timestamp / bucket_size_secs) * bucket_size_secs; + let bucket_key = (payment.prev_channel_id, payment.next_channel_id, bucket_start); + // The marker confirms that every analytics write for this forward succeeded. Wait for + // the whole bucket if any marker is absent, so a later replay cannot add a detail after + // the bucket totals are committed. + if !replay_marker_store.contains_key(&payment.id).await? { + incomplete_buckets.insert(bucket_key); + bucket_groups.remove(&bucket_key); + continue; + } + if incomplete_buckets.contains(&bucket_key) { + continue; + } + bucket_groups.entry(bucket_key).or_default().push(payment); + } + let Some(next_page_token) = page.next_page_token else { + break; + }; + page_token = Some(next_page_token); + } + if bucket_groups.is_empty() { + log_debug!(logger, "No forwarded payments in closed aggregation buckets found"); + return Ok((0, 0)); + } + + 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 = channel_pair_stats_id(&prev_channel_id, &next_channel_id, bucket_start); + + if channel_pair_stats_store.get(&pair_id).await?.is_none() { + 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, one at a time. + 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 result is computed here rather than read from the node, so the four fields that describe a +/// stored bucket do not carry their usual meaning: +/// +/// - [`bucket_start_timestamp`] and [`bucket_size_secs`] span the earliest input bucket start +/// through the latest input bucket end, gaps included, instead of describing one fixed-width +/// bucket. A gap is a period in which this channel pair forwarded nothing, so no bucket was +/// stored for it. +/// - `id` is the key of the earliest input bucket, because that is the bucket the span starts at. +/// It does not identify this result, and writing the result back under it would overwrite that +/// real bucket with a differently-shaped one. +/// - [`aggregated_at_timestamp`] is when this call ran, not when any input was aggregated. +/// +/// Returns `None` if `buckets` is empty or contains statistics for different channel pairs. +/// +/// [`bucket_start_timestamp`]: crate::payment::ChannelPairForwardingStats::bucket_start_timestamp +/// [`bucket_size_secs`]: crate::payment::ChannelPairForwardingStats::bucket_size_secs +/// [`aggregated_at_timestamp`]: crate::payment::ChannelPairForwardingStats::aggregated_at_timestamp +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 now = + SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or(Duration::from_secs(0)).as_secs(); + Some(ChannelPairForwardingStats { + id: channel_pair_stats_id( + &first.prev_channel_id, + &first.next_channel_id, + earliest_bucket_start, + ), + prev_channel_id: first.prev_channel_id, + next_channel_id: first.next_channel_id, + bucket_start_timestamp: earliest_bucket_start, + bucket_size_secs: latest_bucket_end.saturating_sub(earliest_bucket_start), + 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 lightning::util::ser::{Readable, Writeable}; + + use super::*; + use crate::data_store::KeepNoEntries; + #[cfg(feature = "storage-sqlite")] + use crate::io::sqlite_store::SqliteStore; + #[cfg(feature = "storage-sqlite")] + use crate::io::test_utils::random_storage_path; + use crate::io::test_utils::InMemoryStore; + use crate::types::{DynStore, DynStoreWrapper}; + + type TestForwardedPaymentStore = ForwardedPaymentStore; + type TestReplayMarkerStore = ForwardedPaymentReplayMarkerStore; + type TestChannelPairStatsStore = ChannelPairForwardingStatsStore; + + fn test_stores( + ) -> (TestForwardedPaymentStore, TestReplayMarkerStore, TestChannelPairStatsStore, Arc) + { + let (forwarded_payment_store, replay_marker_store, channel_pair_stats_store, logger, _) = + test_stores_with_kv(); + (forwarded_payment_store, replay_marker_store, channel_pair_stats_store, logger) + } + + fn test_stores_with_kv() -> ( + TestForwardedPaymentStore, + TestReplayMarkerStore, + TestChannelPairStatsStore, + Arc, + Arc, + ) { + let kv_store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let logger = Arc::new(Logger::new_log_facade()); + let primary_namespace = "test_forwarded_payments"; + let forwarded_payment_store = ForwardedPaymentStore::new( + Vec::new(), + KeepNoEntries, + primary_namespace.to_string(), + "details".to_string(), + Arc::clone(&kv_store), + Arc::clone(&logger), + ); + let replay_marker_store = ForwardedPaymentReplayMarkerStore::new( + Vec::new(), + KeepNoEntries, + primary_namespace.to_string(), + "replay_markers".to_string(), + Arc::clone(&kv_store), + Arc::clone(&logger), + ); + let channel_pair_stats_store = ChannelPairForwardingStatsStore::new( + Vec::new(), + KeepNoEntries, + primary_namespace.to_string(), + "pair_stats".to_string(), + Arc::clone(&kv_store), + Arc::clone(&logger), + ); + (forwarded_payment_store, replay_marker_store, channel_pair_stats_store, logger, kv_store) + } + + async fn insert_completed_payment( + forwarded_payment_store: &TestForwardedPaymentStore, + replay_marker_store: &TestReplayMarkerStore, payment: ForwardedPaymentDetails, + ) { + let id = payment.id(); + forwarded_payment_store.insert(payment).await.unwrap(); + replay_marker_store.insert(ForwardedPaymentReplayMarker { id }).await.unwrap(); + } + + 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: channel_pair_stats_id(&prev_channel_id, &next_channel_id, bucket_start_timestamp), + 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 = channel_pair_stats_id(&ChannelId([1; 32]), &ChannelId([2; 32]), 42); + let other_id = channel_pair_stats_id(&ChannelId([1; 32]), &ChannelId([2; 32]), 43); + let reversed_id = channel_pair_stats_id(&ChannelId([2; 32]), &ChannelId([1; 32]), 42); + + let key = id.encode_to_hex_str(); + assert_eq!(key.len(), 64); + assert!(key.len() <= KVSTORE_NAMESPACE_KEY_MAX_LEN); + assert!(key.chars().all(|c| KVSTORE_NAMESPACE_KEY_ALPHABET.contains(c))); + assert_ne!(key, other_id.encode_to_hex_str()); + assert_ne!(key, reversed_id.encode_to_hex_str()); + assert_eq!(ChannelPairForwardingStatsId::from_str(&key), Ok(id)); + assert!(ChannelPairForwardingStatsId::decode_from_hex_str("00").is_none()); + assert_eq!(ChannelPairForwardingStatsId::from_str(&key.to_uppercase()), Ok(id)); + } + + #[tokio::test] + async fn aggregation_retains_current_and_previous_buckets() { + let (forwarded_payment_store, replay_marker_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); + insert_completed_payment( + &forwarded_payment_store, + &replay_marker_store, + closed_bucket_payment.clone(), + ) + .await; + insert_completed_payment( + &forwarded_payment_store, + &replay_marker_store, + partial_bucket_payment.clone(), + ) + .await; + + // 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, + &replay_marker_store, + &channel_pair_stats_store, + 60, + 60, + 1_000, + &logger, + ) + .await, + Ok((1, 1)) + ); + + assert!(forwarded_payment_store.get(&closed_bucket_payment.id()).await.unwrap().is_none()); + assert_eq!( + forwarded_payment_store.get(&partial_bucket_payment.id()).await.unwrap(), + Some(partial_bucket_payment) + ); + let bucket_id = channel_pair_stats_id(&ChannelId([1; 32]), &ChannelId([2; 32]), 840); + let stats = channel_pair_stats_store.get(&bucket_id).await.unwrap().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_waits_for_forward_replay_marker() { + let (forwarded_payment_store, replay_marker_store, channel_pair_stats_store, logger) = + test_stores(); + let payment = forwarded_payment(1, 850, 110, 100, 10); + let completed_payment = forwarded_payment(2, 851, 220, 200, 20); + forwarded_payment_store.insert(payment.clone()).await.unwrap(); + insert_completed_payment( + &forwarded_payment_store, + &replay_marker_store, + completed_payment.clone(), + ) + .await; + + assert_eq!( + aggregate_expired_forwarded_payments_at( + &forwarded_payment_store, + &replay_marker_store, + &channel_pair_stats_store, + 60, + 60, + 1_000, + &logger, + ) + .await, + Ok((0, 0)) + ); + assert_eq!( + forwarded_payment_store.get(&payment.id()).await.unwrap(), + Some(payment.clone()) + ); + assert_eq!( + forwarded_payment_store.get(&completed_payment.id()).await.unwrap(), + Some(completed_payment) + ); + + replay_marker_store + .insert(ForwardedPaymentReplayMarker { id: payment.id() }) + .await + .unwrap(); + assert_eq!( + aggregate_expired_forwarded_payments_at( + &forwarded_payment_store, + &replay_marker_store, + &channel_pair_stats_store, + 60, + 60, + 1_000, + &logger, + ) + .await, + Ok((1, 2)) + ); + assert!(forwarded_payment_store.get(&payment.id()).await.unwrap().is_none()); + let bucket_id = + channel_pair_stats_id(&payment.prev_channel_id, &payment.next_channel_id, 840); + assert_eq!( + channel_pair_stats_store.get(&bucket_id).await.unwrap().unwrap().payment_count, + 2 + ); + } + + #[tokio::test] + async fn background_aggregation_runs_immediately() { + let kv_store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let logger = Arc::new(Logger::new_log_facade()); + let forwarding_store = Arc::new(ForwardingStore::new( + Vec::new(), + ForwardedPaymentTrackingMode::Stats, + kv_store, + Arc::clone(&logger), + )); + let payment = forwarded_payment(1, 1, 110, 100, 10); + forwarding_store.details.insert(payment.clone()).await.unwrap(); + forwarding_store + .replay_markers + .insert(ForwardedPaymentReplayMarker { id: payment.id() }) + .await + .unwrap(); + let (_stop_sender, stop_receiver) = tokio::sync::watch::channel(()); + + tokio::time::timeout( + Duration::from_secs(1), + run_forwarded_payment_aggregation(stop_receiver, Arc::clone(&forwarding_store), 0), + ) + .await + .unwrap(); + + assert!(forwarding_store.details.is_empty().await.unwrap()); + let bucket_id = + channel_pair_stats_id(&payment.prev_channel_id, &payment.next_channel_id, 0); + assert_eq!( + forwarding_store + .channel_pair_stats + .get(&bucket_id) + .await + .unwrap() + .unwrap() + .payment_count, + 1 + ); + } + + #[tokio::test] + async fn forwarding_store_records_details_and_channel_stats() { + let kv_store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let logger = Arc::new(Logger::new_log_facade()); + let forwarding_store = ForwardingStore::new( + Vec::new(), + ForwardedPaymentTrackingMode::Detailed, + kv_store, + logger, + ); + let prev_channel_id = ChannelId([1; 32]); + let next_channel_id = ChannelId([2; 32]); + let prev_htlcs = [HTLCLocator { + channel_id: prev_channel_id, + htlc_id: Some(7), + amount_msat: Some(110), + user_channel_id: Some(3), + node_id: None, + }]; + let next_htlcs = [HTLCLocator { + channel_id: next_channel_id, + htlc_id: Some(8), + amount_msat: Some(100), + user_channel_id: Some(4), + node_id: None, + }]; + let details_id = forwarded_payment_id(&prev_channel_id, 7); + + let forward = || ForwardRecord { + prev_htlcs: &prev_htlcs, + next_htlcs: &next_htlcs, + total_fee_earned_msat: Some(10), + skimmed_fee_msat: Some(2), + claim_from_onchain_tx: true, + outbound_amount_forwarded_msat: 100, + }; + forwarding_store.record_forward(forward()).await.unwrap(); + forwarding_store.record_forward(forward()).await.unwrap(); + + let details = forwarding_store.payment(&details_id).await.unwrap().unwrap(); + assert_eq!(details.inbound_amount_forwarded_msat, Some(110)); + assert_eq!(details.outbound_amount_forwarded_msat, Some(100)); + assert_eq!(details.total_fee_earned_msat, Some(10)); + assert_eq!(details.skimmed_fee_msat, Some(2)); + assert_eq!(ForwardedPaymentId::from_str(&details_id.to_string()), Ok(details_id)); + assert_eq!( + ForwardedPaymentId::from_str("not-an-id"), + Err(Error::InvalidForwardedPaymentId) + ); + assert_eq!(ForwardedPaymentId::from_str("00"), Err(Error::InvalidForwardedPaymentId)); + assert_eq!( + ForwardedPaymentId::from_str(&details_id.to_string().to_uppercase()), + Ok(details_id) + ); + + let inbound = forwarding_store.channel_stats(&prev_channel_id).await.unwrap().unwrap(); + assert_eq!(inbound.inbound_payments_forwarded, 1); + assert_eq!(inbound.total_inbound_amount_msat, 110); + assert_eq!(inbound.total_fee_earned_msat, Some(10)); + assert_eq!(inbound.total_skimmed_fee_msat, 2); + + let outbound = forwarding_store.channel_stats(&next_channel_id).await.unwrap().unwrap(); + assert_eq!(outbound.outbound_payments_forwarded, 1); + assert_eq!(outbound.total_outbound_amount_msat, 100); + assert_eq!(outbound.total_fee_earned_msat, Some(0)); + assert_eq!(outbound.onchain_claims_count, 1); + assert_eq!(forwarding_store.list_payments(None).await.unwrap().objects.len(), 1); + } + + #[tokio::test] + async fn forwarding_store_deduplicates_stats_mode_replays() { + let kv_store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let logger = Arc::new(Logger::new_log_facade()); + let forwarding_store = + ForwardingStore::new(Vec::new(), ForwardedPaymentTrackingMode::Stats, kv_store, logger); + let prev_channel_id = ChannelId([1; 32]); + let next_channel_id = ChannelId([2; 32]); + let prev_htlcs = [HTLCLocator { + channel_id: prev_channel_id, + htlc_id: Some(7), + amount_msat: Some(110), + user_channel_id: Some(3), + node_id: None, + }]; + let next_htlcs = [HTLCLocator { + channel_id: next_channel_id, + htlc_id: Some(8), + amount_msat: Some(100), + user_channel_id: Some(4), + node_id: None, + }]; + let forward = || ForwardRecord { + prev_htlcs: &prev_htlcs, + next_htlcs: &next_htlcs, + total_fee_earned_msat: Some(10), + skimmed_fee_msat: Some(2), + claim_from_onchain_tx: false, + outbound_amount_forwarded_msat: 100, + }; + + forwarding_store.record_forward(forward()).await.unwrap(); + forwarding_store.record_forward(forward()).await.unwrap(); + + let inbound = forwarding_store.channel_stats(&prev_channel_id).await.unwrap().unwrap(); + assert_eq!(inbound.inbound_payments_forwarded, 1); + assert_eq!(inbound.total_inbound_amount_msat, 110); + assert_eq!(inbound.total_fee_earned_msat, Some(10)); + let outbound = forwarding_store.channel_stats(&next_channel_id).await.unwrap().unwrap(); + assert_eq!(outbound.outbound_payments_forwarded, 1); + assert_eq!(outbound.total_outbound_amount_msat, 100); + assert!(forwarding_store.list_payments(None).await.unwrap().objects.is_empty()); + } + + #[tokio::test] + async fn forwarding_store_skips_ineligible_forwards() { + let kv_store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let logger = Arc::new(Logger::new_log_facade()); + let forwarding_store = ForwardingStore::new( + Vec::new(), + ForwardedPaymentTrackingMode::Detailed, + kv_store, + logger, + ); + let prev_htlcs = [ + HTLCLocator { + channel_id: ChannelId([1; 32]), + htlc_id: None, + amount_msat: Some(110), + user_channel_id: None, + node_id: None, + }, + HTLCLocator { + channel_id: ChannelId([2; 32]), + htlc_id: None, + amount_msat: Some(110), + user_channel_id: None, + node_id: None, + }, + ]; + let next_htlcs = [HTLCLocator { + channel_id: ChannelId([3; 32]), + htlc_id: None, + amount_msat: Some(100), + user_channel_id: None, + node_id: None, + }]; + + forwarding_store + .record_forward(ForwardRecord { + prev_htlcs: &prev_htlcs, + next_htlcs: &next_htlcs, + total_fee_earned_msat: Some(10), + skimmed_fee_msat: None, + claim_from_onchain_tx: false, + outbound_amount_forwarded_msat: 100, + }) + .await + .unwrap(); + forwarding_store + .record_forward(ForwardRecord { + prev_htlcs: &prev_htlcs[..1], + next_htlcs: &next_htlcs, + total_fee_earned_msat: None, + skimmed_fee_msat: None, + claim_from_onchain_tx: true, + outbound_amount_forwarded_msat: 100, + }) + .await + .unwrap(); + + assert!(forwarding_store.details.is_empty().await.unwrap()); + assert!(forwarding_store.channel_stats.is_empty().await.unwrap()); + } + + #[tokio::test] + async fn aggregation_preserves_unknown_fees() { + let (forwarded_payment_store, replay_marker_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; + insert_completed_payment(&forwarded_payment_store, &replay_marker_store, known_fee_payment) + .await; + insert_completed_payment( + &forwarded_payment_store, + &replay_marker_store, + unknown_fee_payment, + ) + .await; + + assert_eq!( + aggregate_expired_forwarded_payments_at( + &forwarded_payment_store, + &replay_marker_store, + &channel_pair_stats_store, + 60, + 60, + 1_000, + &logger, + ) + .await, + Ok((1, 2)) + ); + + let bucket_id = channel_pair_stats_id(&ChannelId([1; 32]), &ChannelId([2; 32]), 840); + let stats = channel_pair_stats_store.get(&bucket_id).await.unwrap().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 = StoredChannelForwardingStats::from(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, + last_inbound_forward_id: None, + last_outbound_forward_id: None, + })); + assert_eq!(stats.total_fee_earned_msat, None); + } + + #[test] + fn stored_channel_stats_reads_legacy_stats() { + let legacy = ChannelForwardingStats { + channel_id: ChannelId([1; 32]), + counterparty_node_id: None, + inbound_payments_forwarded: 1, + outbound_payments_forwarded: 2, + total_inbound_amount_msat: 110, + total_outbound_amount_msat: 200, + total_fee_earned_msat: Some(10), + total_skimmed_fee_msat: 2, + onchain_claims_count: 1, + first_forwarded_at_timestamp: 850, + last_forwarded_at_timestamp: 851, + }; + + let encoded = legacy.encode(); + let stored = StoredChannelForwardingStats::read(&mut &encoded[..]).unwrap(); + assert_eq!(stored.last_inbound_forward_id, None); + assert_eq!(stored.last_outbound_forward_id, None); + assert_eq!(ChannelForwardingStats::from(stored), legacy); + } + + #[tokio::test] + async fn aggregation_keeps_both_retained_bucket_boundaries() { + let (forwarded_payment_store, replay_marker_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); + insert_completed_payment( + &forwarded_payment_store, + &replay_marker_store, + older_bucket_payment.clone(), + ) + .await; + insert_completed_payment( + &forwarded_payment_store, + &replay_marker_store, + previous_bucket_payment.clone(), + ) + .await; + insert_completed_payment( + &forwarded_payment_store, + &replay_marker_store, + current_bucket_payment.clone(), + ) + .await; + + assert_eq!( + aggregate_expired_forwarded_payments_at( + &forwarded_payment_store, + &replay_marker_store, + &channel_pair_stats_store, + 60, + 60, + 900, + &logger, + ) + .await, + Ok((1, 1)) + ); + + assert!(forwarded_payment_store.get(&older_bucket_payment.id()).await.unwrap().is_none()); + assert_eq!( + forwarded_payment_store.get(&previous_bucket_payment.id()).await.unwrap(), + Some(previous_bucket_payment) + ); + assert_eq!( + forwarded_payment_store.get(¤t_bucket_payment.id()).await.unwrap(), + Some(current_bucket_payment) + ); + } + + #[tokio::test] + async fn zero_retention_cleans_up_after_the_current_bucket_closes() { + let (forwarded_payment_store, replay_marker_store, channel_pair_stats_store, logger) = + test_stores(); + let payment = forwarded_payment(1, 899, 110, 100, 10); + insert_completed_payment(&forwarded_payment_store, &replay_marker_store, payment.clone()) + .await; + + assert_eq!( + aggregate_expired_forwarded_payments_at( + &forwarded_payment_store, + &replay_marker_store, + &channel_pair_stats_store, + 60, + 0, + 899, + &logger, + ) + .await, + Ok((0, 0)) + ); + assert_eq!( + forwarded_payment_store.get(&payment.id()).await.unwrap(), + Some(payment.clone()) + ); + + assert_eq!( + aggregate_expired_forwarded_payments_at( + &forwarded_payment_store, + &replay_marker_store, + &channel_pair_stats_store, + 60, + 0, + 900, + &logger, + ) + .await, + Ok((1, 1)) + ); + assert!(forwarded_payment_store.get(&payment.id()).await.unwrap().is_none()); + } + + #[tokio::test] + async fn aggregation_retry_only_cleans_up_committed_bucket() { + let (forwarded_payment_store, replay_marker_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(); + insert_completed_payment( + &forwarded_payment_store, + &replay_marker_store, + remaining_payment.clone(), + ) + .await; + + // 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, + &replay_marker_store, + &channel_pair_stats_store, + 60, + 60, + 1_000, + &logger, + ) + .await, + Ok((0, 1)) + ); + + assert!(forwarded_payment_store.get(&remaining_payment.id()).await.unwrap().is_none()); + assert_eq!( + channel_pair_stats_store.get(&committed_stats.id()).await.unwrap(), + Some(committed_stats) + ); + } + + #[tokio::test] + async fn aggregation_preserves_details_if_commit_marker_is_corrupt() { + let ( + forwarded_payment_store, + replay_marker_store, + channel_pair_stats_store, + logger, + kv_store, + ) = test_stores_with_kv(); + let payment = forwarded_payment(2, 851, 220, 200, 20); + let pair_id = + channel_pair_stats_id(&payment.prev_channel_id, &payment.next_channel_id, 840); + insert_completed_payment(&forwarded_payment_store, &replay_marker_store, payment.clone()) + .await; + KVStore::write( + &*kv_store, + "test_forwarded_payments", + "pair_stats", + &pair_id.encode_to_hex_str(), + vec![0xff], + ) + .await + .unwrap(); + + assert_eq!( + aggregate_expired_forwarded_payments_at( + &forwarded_payment_store, + &replay_marker_store, + &channel_pair_stats_store, + 60, + 60, + 1_000, + &logger, + ) + .await, + Err(Error::PersistenceFailed) + ); + assert_eq!(forwarded_payment_store.get(&payment.id()).await.unwrap(), Some(payment)); + } + + #[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.payment_count, 3); + assert_eq!(cumulative.total_fee_earned_msat, Some(40)); + // The span runs from the earliest bucket start to the latest bucket end, gaps included. + assert_eq!(cumulative.bucket_start_timestamp, 840); + assert_eq!(cumulative.bucket_size_secs, 180); + } + + #[test] + fn cumulative_stats_reject_mismatched_channel_pairs() { + let mut other_pair = channel_pair_stats(900, 60, 1_000, 1, 110, 100, 10, 902, 902); + other_pair.next_channel_id = ChannelId([3; 32]); + + assert!(aggregate_channel_pair_stats(&[]).is_none()); + assert!(aggregate_channel_pair_stats(&[ + channel_pair_stats(840, 60, 950, 2, 330, 300, 30, 850, 851), + other_pair, + ]) + .is_none()); + } + + #[tokio::test] + async fn retention_does_not_change_bucket_geometry() { + let (forwarded_payment_store, replay_marker_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(); + insert_completed_payment(&forwarded_payment_store, &replay_marker_store, payment.clone()) + .await; + + assert_eq!( + aggregate_expired_forwarded_payments_at( + &forwarded_payment_store, + &replay_marker_store, + &channel_pair_stats_store, + 3_600, + 7_200, + 15_000, + &logger, + ) + .await, + Ok((1, 1)) + ); + + let new_id = + channel_pair_stats_id(&payment.prev_channel_id, &payment.next_channel_id, 3_600); + assert_eq!(channel_pair_stats_store.get(&old_stats.id()).await.unwrap(), Some(old_stats)); + assert_eq!(channel_pair_stats_store.get(&new_id).await.unwrap().unwrap().payment_count, 1); + assert!(forwarded_payment_store.get(&payment.id()).await.unwrap().is_none()); + } + + #[tokio::test] + #[cfg(feature = "storage-sqlite")] + 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 secondary_namespace = "pair_stats"; + let stats_store = ChannelPairForwardingStatsStore::new( + Vec::new(), + KeepNoEntries, + namespace.to_string(), + secondary_namespace.to_string(), + 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, secondary_namespace).await.unwrap(); + assert_eq!(keys, vec![stats.id.to_string()]); + let bytes = + KVStore::read(&*kv_store, namespace, secondary_namespace, &keys[0]).await.unwrap(); + assert_eq!(ChannelPairForwardingStats::read(&mut &bytes[..]).unwrap(), stats); + } + + #[tokio::test] + async fn uncached_data_store_reads_existing_objects_across_pages() { + let kv_store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let logger = Arc::new(Logger::new_log_facade()); + let namespace = "paged_forwarded_payments"; + let store = ForwardedPaymentStore::new( + Vec::new(), + KeepNoEntries, + namespace.to_string(), + "details".to_string(), + Arc::clone(&kv_store), + Arc::clone(&logger), + ); + for id in 0..=50 { + store.insert(forwarded_payment(id, id as u64, 110, 100, 10)).await.unwrap(); + } + assert_eq!(store.cached_len(), 0); + + // Recreate the store to verify it has no in-memory state to preload. + let reopened_store = ForwardedPaymentStore::new( + Vec::new(), + KeepNoEntries, + namespace.to_string(), + "details".to_string(), + kv_store, + logger, + ); + let oldest_payment = forwarded_payment(0, 0, 110, 100, 10); + assert_eq!( + reopened_store.get(&oldest_payment.id()).await.unwrap(), + Some(oldest_payment.clone()) + ); + assert_eq!(reopened_store.cached_len(), 0); + + let first_page = reopened_store.list_page(None).await.unwrap(); + assert_eq!(first_page.objects.len(), 50); + let second_page = reopened_store.list_page(first_page.next_page_token).await.unwrap(); + assert_eq!(second_page.objects, vec![oldest_payment.clone()]); + assert!(second_page.next_page_token.is_none()); + } +} diff --git a/src/payment/mod.rs b/src/payment/mod.rs index 13dbe5106..e1c60da79 100644 --- a/src/payment/mod.rs +++ b/src/payment/mod.rs @@ -10,6 +10,8 @@ pub(crate) mod asynchronous; mod bolt11; mod bolt12; +mod forwarding; +pub(crate) mod forwarding_store; #[cfg(feature = "unified-payments")] mod hrn; mod onchain; @@ -22,6 +24,11 @@ mod unified; pub use bolt11::Bolt11Payment; pub(crate) use bolt11::PaymentMetadata; pub use bolt12::{Bolt12Payment, PayerProofOptions}; +pub use forwarding::{ + ChannelForwardingStats, ChannelForwardingStatsPage, ChannelPairForwardingStats, + ChannelPairForwardingStatsId, ChannelPairForwardingStatsPage, ForwardedPaymentDetails, + ForwardedPaymentDetailsPage, ForwardedPaymentId, ForwardingAnalytics, +}; #[cfg(feature = "unified-payments")] pub(crate) use hrn::HRNResolver; pub use onchain::OnchainPayment; diff --git a/src/payment/pending_payment_store.rs b/src/payment/pending_payment_store.rs index 30a113537..e14f64c38 100644 --- a/src/payment/pending_payment_store.rs +++ b/src/payment/pending_payment_store.rs @@ -9,7 +9,7 @@ use bitcoin::Txid; use lightning::impl_writeable_tlv_based; use lightning::ln::channelmanager::PaymentId; -use crate::data_store::{StorableObject, StorableObjectUpdate}; +use crate::data_store::{StorableObject, StorableObjectUpdate, UpdatableObject}; use crate::payment::store::PaymentDetailsUpdate; use crate::payment::{PaymentDetails, PaymentKind}; @@ -78,11 +78,14 @@ pub(crate) struct PendingPaymentDetailsUpdate { impl StorableObject for PendingPaymentDetails { type Id = PaymentId; - type Update = PendingPaymentDetailsUpdate; fn id(&self) -> Self::Id { self.details.id } +} + +impl UpdatableObject for PendingPaymentDetails { + type Update = PendingPaymentDetailsUpdate; fn update(&mut self, update: Self::Update) -> bool { let mut updated = false; diff --git a/src/payment/store.rs b/src/payment/store.rs index 3163ed15b..41c39045f 100644 --- a/src/payment/store.rs +++ b/src/payment/store.rs @@ -22,7 +22,7 @@ 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::{StorableObject, StorableObjectId, StorableObjectUpdate, UpdatableObject}; use crate::hex_utils; /// An opaque token used to continue a paginated listing. @@ -192,11 +192,14 @@ impl StorableObjectId for PaymentId { } impl StorableObject for PaymentDetails { type Id = PaymentId; - type Update = PaymentDetailsUpdate; fn id(&self) -> Self::Id { self.id } +} + +impl UpdatableObject for PaymentDetails { + type Update = PaymentDetailsUpdate; fn update(&mut self, update: Self::Update) -> bool { debug_assert_eq!( diff --git a/src/types.rs b/src/types.rs index 1a61daa10..fd86d1bcd 100644 --- a/src/types.rs +++ b/src/types.rs @@ -37,12 +37,14 @@ use lightning_types::features::ChannelTypeFeatures; use crate::chain::ChainSource; use crate::config::{AnchorChannelsConfig, ChannelConfig}; -use crate::data_store::{DataStore, KeepAllEntries, KeepLeastRecentlyUsed}; +use crate::data_store::{DataStore, KeepAllEntries, KeepLeastRecentlyUsed, KeepNoEntries}; use crate::fee_estimator::OnchainFeeEstimator; use crate::ffi::maybe_wrap; use crate::logger::Logger; use crate::message_handler::NodeCustomMessageHandler; -use crate::payment::{PaymentDetails, PendingPaymentDetails}; +use crate::payment::{ + ChannelPairForwardingStats, ForwardedPaymentDetails, PaymentDetails, PendingPaymentDetails, +}; use crate::runtime::RuntimeSpawner; #[cfg(feature = "uniffi")] @@ -333,6 +335,12 @@ pub(crate) type BumpTransactionEventHandler = >; pub(crate) type PaymentStore = DataStore, KeepLeastRecentlyUsed>; +pub(crate) type ForwardedPaymentStore = + DataStore, KeepNoEntries>; +pub(crate) type ChannelForwardingStatsStore = + DataStore>; +pub(crate) type ChannelPairForwardingStatsStore = + DataStore, KeepNoEntries>; /// A local, potentially user-provided, identifier of a channel. /// diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index b9c12b4a7..48d6b17a6 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -54,7 +54,7 @@ use lightning_invoice::RawBolt11Invoice; use persist::KVStoreWalletPersister; use crate::config::{Config, ADDRESS_POOL_SIZE}; -use crate::data_store::StorableObject; +use crate::data_store::UpdatableObject; #[cfg(test)] use crate::data_store::{KeepAllEntries, KeepLeastRecentlyUsed}; use crate::fee_estimator::{ConfirmationTarget, FeeEstimator, OnchainFeeEstimator}; diff --git a/tests/integration_tests_rust.rs b/tests/integration_tests_rust.rs index 0dad32d6a..457ed2309 100644 --- a/tests/integration_tests_rust.rs +++ b/tests/integration_tests_rust.rs @@ -35,13 +35,14 @@ use common::{ use electrsd::corepc_node::{self, Node as BitcoinD}; use electrsd::ElectrsD; use ldk_node::config::{ - AsyncPaymentsRole, EsploraSyncConfig, ADDRESS_POOL_SIZE, DEFAULT_FULL_SCAN_STOP_GAP, + AsyncPaymentsRole, EsploraSyncConfig, ForwardedPaymentTrackingMode, ADDRESS_POOL_SIZE, + DEFAULT_FULL_SCAN_STOP_GAP, }; use ldk_node::entropy::NodeEntropy; use ldk_node::liquidity::LSPS2ServiceConfig; use ldk_node::payment::{ - ConfirmationStatus, PayerProofOptions, PaymentDetails, PaymentDirection, PaymentKind, - PaymentStatus, TransactionType, UnifiedPaymentResult, + ConfirmationStatus, ForwardedPaymentId, PayerProofOptions, PaymentDetails, PaymentDirection, + PaymentKind, PaymentStatus, TransactionType, UnifiedPaymentResult, }; use ldk_node::{BuildError, Builder, Event, Node, NodeError, ReserveType}; use lightning::ln::channelmanager::PaymentId; @@ -731,6 +732,152 @@ async fn multi_hop_sending() { expect_payment_received_event!(&nodes[4], 2_500_000); let fee_paid_msat = Some(2000); expect_payment_successful_event!(nodes[0], outbound_payment_id, Some(fee_paid_msat)); + + // N1 forwarded the payment, so it records the forward against both of its channels: the one it + // received on and the one it sent on. N0 only sent, so it records nothing. + let forwarding = nodes[1].forwarding_analytics(); + assert_eq!(forwarding.tracking_mode(), ForwardedPaymentTrackingMode::Stats); + + let inbound_channel_id = nodes[1] + .list_channels() + .iter() + .find(|c| c.counterparty.node_id == nodes[0].node_id()) + .unwrap() + .channel_id; + let inbound_stats = forwarding.channel_stats(&inbound_channel_id).unwrap().unwrap(); + assert_eq!(inbound_stats.inbound_payments_forwarded, 1); + assert_eq!(inbound_stats.outbound_payments_forwarded, 0); + assert_eq!(inbound_stats.counterparty_node_id, Some(nodes[0].node_id())); + // Fees are attributed to the incoming channel, so all of N1's fee lands here. + let fee_earned_msat = inbound_stats.total_fee_earned_msat.unwrap(); + assert!(fee_earned_msat > 0); + assert_eq!(inbound_stats.total_outbound_amount_msat, 0); + + // Exactly one of N1's two outgoing channels carried the forward, since the payment took either + // the N2 or the N3 route. + let outbound_stats = forwarding + .list_channel_stats(None) + .unwrap() + .stats + .into_iter() + .filter(|s| s.outbound_payments_forwarded > 0) + .collect::>(); + assert_eq!(outbound_stats.len(), 1); + assert_eq!(outbound_stats[0].outbound_payments_forwarded, 1); + assert_eq!(outbound_stats[0].inbound_payments_forwarded, 0); + assert_ne!(outbound_stats[0].channel_id, inbound_channel_id); + + // N1 sends on more than the recipient gets, because the next hop takes a fee of its own too. + // What N1 received is what it sent on plus the fee it kept. + assert!(outbound_stats[0].total_outbound_amount_msat > 2_500_000); + assert_eq!( + inbound_stats.total_inbound_amount_msat, + outbound_stats[0].total_outbound_amount_msat + fee_earned_msat + ); + + // `Stats` is the default mode, so no per-payment detail is kept. + assert!(forwarding.list_payments(None).unwrap().payments.is_empty()); + + assert!(nodes[0].forwarding_analytics().list_channel_stats(None).unwrap().stats.is_empty()); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +async fn detailed_forwarded_payment_tracking() { + let (bitcoind, electrsd) = setup_bitcoind_and_electrsd(); + let chain_source = random_chain_source(&bitcoind, &electrsd); + let node_a = setup_node(&chain_source, random_config()); + let mut router_config = random_config(); + router_config.node_config.forwarded_payment_tracking_mode = + ForwardedPaymentTrackingMode::Detailed; + let node_b = setup_node(&chain_source, router_config); + let node_c = setup_node(&chain_source, random_config()); + + let addr_a = node_a.onchain_payment().new_address().unwrap(); + let addr_b = node_b.onchain_payment().new_address().unwrap(); + let addr_c = node_c.onchain_payment().new_address().unwrap(); + let premine_amount_sat = 5_000_000; + premine_and_distribute_funds( + &bitcoind.client, + &electrsd.client, + vec![addr_a, addr_b, addr_c], + Amount::from_sat(premine_amount_sat), + ) + .await; + + for node in [&node_a, &node_b, &node_c] { + node.sync_wallets().unwrap(); + } + + // A -> B -> C, announced so that A can find the route through B. + open_channel(&node_a, &node_b, 1_000_000, true, &electrsd).await; + open_channel(&node_b, &node_c, 1_000_000, true, &electrsd).await; + + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + + for node in [&node_a, &node_b, &node_c] { + node.sync_wallets().unwrap(); + } + + expect_channel_ready_event!(node_a, node_b.node_id()); + expect_channel_ready_events!(node_b, node_a.node_id(), node_c.node_id()); + expect_channel_ready_event!(node_c, node_b.node_id()); + + // Sleep a bit for gossip to propagate. + tokio::time::sleep(Duration::from_secs(1)).await; + + let forwarding = node_b.forwarding_analytics(); + assert_eq!(forwarding.tracking_mode(), ForwardedPaymentTrackingMode::Detailed); + + let amount_msat = 2_500_000; + let invoice_description = + Bolt11InvoiceDescription::Direct(Description::new(String::from("detailed")).unwrap()); + let invoice = + node_c.bolt11_payment().receive(amount_msat, &invoice_description.into(), 3600).unwrap(); + let payment_id = node_a.bolt11_payment().send(&invoice, None).unwrap(); + + expect_event!(node_b, PaymentForwarded); + expect_payment_received_event!(node_c, amount_msat); + expect_payment_successful_event!(node_a, payment_id, None); + + // In `Detailed` mode the forward is kept as an individual record until its bucket closes. + let page = forwarding.list_payments(None).unwrap(); + assert_eq!(page.payments.len(), 1); + assert!(page.next_page_token.is_none()); + let details = &page.payments[0]; + + let inbound_channel_id = node_b + .list_channels() + .iter() + .find(|c| c.counterparty.node_id == node_a.node_id()) + .unwrap() + .channel_id; + let outbound_channel_id = node_b + .list_channels() + .iter() + .find(|c| c.counterparty.node_id == node_c.node_id()) + .unwrap() + .channel_id; + assert_eq!(details.prev_channel_id, inbound_channel_id); + assert_eq!(details.next_channel_id, outbound_channel_id); + assert_eq!(details.prev_node_id, Some(node_a.node_id())); + assert_eq!(details.next_node_id, Some(node_c.node_id())); + assert_eq!(details.outbound_amount_forwarded_msat, Some(amount_msat)); + assert!(!details.claim_from_onchain_tx); + + let fee_earned_msat = details.total_fee_earned_msat.unwrap(); + assert_eq!(details.inbound_amount_forwarded_msat, Some(amount_msat + fee_earned_msat)); + + // The opaque id round-trips through the single-payment lookup. + assert_eq!(forwarding.payment(&details.id).unwrap().as_ref(), Some(details)); + assert!(forwarding.payment(&ForwardedPaymentId([0; 32])).unwrap().is_none()); + + // Channel statistics are recorded in both modes, so they agree with the detail record. + let inbound_stats = forwarding.channel_stats(&inbound_channel_id).unwrap().unwrap(); + assert_eq!(inbound_stats.inbound_payments_forwarded, 1); + assert_eq!(inbound_stats.total_fee_earned_msat, Some(fee_earned_msat)); + let outbound_stats = forwarding.channel_stats(&outbound_channel_id).unwrap().unwrap(); + assert_eq!(outbound_stats.outbound_payments_forwarded, 1); + assert_eq!(outbound_stats.total_outbound_amount_msat, amount_msat); } #[tokio::test(flavor = "multi_thread", worker_threads = 1)]