From e87f7b29cc73550f1d54dad8e1686e0140c75739 Mon Sep 17 00:00:00 2001 From: benthecarman Date: Wed, 19 Aug 2026 03:37:46 -0500 Subject: [PATCH 1/2] Split merge behavior out of StorableObject StorableObject required an update representation from every type a store holds, but only DataStore merges updates into an existing object. A store that just reads, writes and deletes whole objects still had to supply an Update type, plus update and to_update methods that nothing would ever call. Move those three onto a new UpdatableObject trait, and leave StorableObject with the id and serialization any store needs. DataStore bounds on UpdatableObject, so its behavior is unchanged. Co-Authored-By: Claude Opus 5 (1M context) --- src/data_store.rs | 22 +++++++++++++++++----- src/payment/pending_payment_store.rs | 7 +++++-- src/payment/store.rs | 7 +++++-- src/wallet/mod.rs | 2 +- 4 files changed, 28 insertions(+), 10 deletions(-) diff --git a/src/data_store.rs b/src/data_store.rs index a9fe0d0f5..d26f0f49a 100644 --- a/src/data_store.rs +++ b/src/data_store.rs @@ -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; } @@ -264,7 +273,7 @@ pub(crate) struct DataStorePage { pub next_page_token: Option, } -pub(crate) struct DataStore +pub(crate) struct DataStore where L::Target: LdkLogger, { @@ -282,7 +291,7 @@ where cache_policy: PhantomData

, } -impl DataStore +impl DataStore where L::Target: LdkLogger, { @@ -730,7 +739,7 @@ where } } -impl DataStore +impl DataStore where L::Target: LdkLogger, { @@ -832,11 +841,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; 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/wallet/mod.rs b/src/wallet/mod.rs index 892f40914..d823c6a22 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}; From 975082c42718cc95a37d34157f319facbe0b2f14 Mon Sep 17 00:00:00 2001 From: benthecarman Date: Mon, 27 Jul 2026 13:46:05 -0500 Subject: [PATCH 2/2] Add forwarded payment tracking Store only unambiguous single-HTLC forwarding events. Aggregate events into channel and channel-pair statistics. Use fixed one-hour buckets for detailed records. Expose bucket identifiers as opaque strings. AI-assisted-by: OpenAI Codex --- bindings/ldk_node.udl | 17 + src/builder.rs | 105 ++- src/config.rs | 32 +- src/data_store.rs | 10 + src/event.rs | 170 ++++- src/io/mod.rs | 14 + src/lib.rs | 83 ++- src/payment/forwarding.rs | 403 +++++++++++ src/payment/forwarding_store.rs | 1190 +++++++++++++++++++++++++++++++ src/payment/mod.rs | 7 + src/types.rs | 9 +- tests/integration_tests_rust.rs | 149 +++- 12 files changed, 2157 insertions(+), 32 deletions(-) create mode 100644 src/payment/forwarding.rs create mode 100644 src/payment/forwarding_store.rs diff --git a/bindings/ldk_node.udl b/bindings/ldk_node.udl index fddf5940c..4d69a2132 100644 --- a/bindings/ldk_node.udl +++ b/bindings/ldk_node.udl @@ -11,6 +11,8 @@ typedef dictionary ElectrumSyncConfig; typedef dictionary TorConfig; +typedef enum ForwardedPaymentTrackingMode; + typedef interface NodeEntropy; typedef interface ProbingConfig; @@ -113,6 +115,7 @@ interface Node { OnchainPayment onchain_payment(); UnifiedPayment unified_payment(); Liquidity liquidity(); + Forwarding forwarding(); [Throws=NodeError] void lnurl_auth(string lnurl); [Throws=NodeError] @@ -186,6 +189,8 @@ typedef interface UnifiedPayment; typedef interface Liquidity; +typedef interface Forwarding; + [Error] enum NodeError { "AlreadyRunning", @@ -442,3 +447,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 f0f38783f..8a6dc795c 100644 --- a/src/builder.rs +++ b/src/builder.rs @@ -67,7 +67,13 @@ use crate::io::utils::{ }; use crate::io::vss_store::VssStoreBuilder; use crate::io::{ - self, PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, + self, CHANNEL_FORWARDING_STATS_PERSISTENCE_PRIMARY_NAMESPACE, + CHANNEL_FORWARDING_STATS_PERSISTENCE_SECONDARY_NAMESPACE, + CHANNEL_PAIR_FORWARDING_STATS_PERSISTENCE_PRIMARY_NAMESPACE, + CHANNEL_PAIR_FORWARDING_STATS_PERSISTENCE_SECONDARY_NAMESPACE, + FORWARDED_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, + FORWARDED_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, + PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, PENDING_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, }; @@ -84,7 +90,8 @@ use crate::probing::{ use crate::runtime::{Runtime, RuntimeSpawner}; use crate::tx_broadcaster::TransactionBroadcaster; use crate::types::{ - AsyncPersister, ChainMonitor, ChannelManager, DynStore, DynStoreRef, DynStoreWrapper, + AsyncPersister, ChainMonitor, ChannelForwardingStatsStore, ChannelManager, + ChannelPairForwardingStatsStore, DynStore, DynStoreRef, DynStoreWrapper, ForwardedPaymentStore, GossipSync, Graph, HRNResolver, KeysManager, MessageRouter, OnionMessenger, PaymentStore, PeerManager, PendingPaymentStore, }; @@ -1457,26 +1464,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, + CHANNEL_FORWARDING_STATS_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 { @@ -1509,6 +1527,35 @@ fn build_with_store_internal( }, }; + let forwarded_payment_store = Arc::new(ForwardedPaymentStore::new( + FORWARDED_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE.to_string(), + FORWARDED_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE.to_string(), + Arc::clone(&kv_store), + Arc::clone(&logger), + )); + + let channel_forwarding_stats_store = match channel_forwarding_stats_res { + Ok(stats) => Arc::new(ChannelForwardingStatsStore::new( + stats, + KeepAllEntries, + CHANNEL_FORWARDING_STATS_PERSISTENCE_PRIMARY_NAMESPACE.to_string(), + CHANNEL_FORWARDING_STATS_PERSISTENCE_SECONDARY_NAMESPACE.to_string(), + Arc::clone(&kv_store), + Arc::clone(&logger), + )), + Err(e) => { + log_error!(logger, "Failed to read channel forwarding stats from store: {}", e); + return Err(BuildError::ReadFailed); + }, + }; + + let channel_pair_forwarding_stats_store = Arc::new(ChannelPairForwardingStatsStore::new( + CHANNEL_PAIR_FORWARDING_STATS_PERSISTENCE_PRIMARY_NAMESPACE.to_string(), + CHANNEL_PAIR_FORWARDING_STATS_PERSISTENCE_SECONDARY_NAMESPACE.to_string(), + Arc::clone(&kv_store), + Arc::clone(&logger), + )); + let (chain_source, chain_tip_opt) = match chain_data_source_config { Some(ChainDataSourceConfig::Esplora { server_url, headers, sync_config }) => { let sync_config = sync_config.unwrap_or(EsploraSyncConfig::default()); @@ -2359,6 +2406,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, @@ -2386,6 +2443,10 @@ fn build_with_store_internal( scorer, peer_store, payment_store, + forwarded_payment_store, + channel_forwarding_stats_store, + channel_pair_forwarding_stats_store, + forwarded_payment_aggregation_retention_secs, lnurl_auth, is_running, node_metrics, diff --git a/src/config.rs b/src/config.rs index a409b9e48..25c6d1975 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. @@ -189,9 +213,10 @@ pub(crate) const LIQUIDITY_DISCOVERY_RETRY_MAX_DELAY: Duration = Duration::from_ /// | `tor_config` | None | /// | `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 { @@ -264,6 +289,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 { @@ -281,6 +308,7 @@ impl Default for Config { node_alias: None, 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 d26f0f49a..19514584f 100644 --- a/src/data_store.rs +++ b/src/data_store.rs @@ -50,6 +50,16 @@ pub(crate) trait StorableObjectId: Clone + std::hash::Hash + PartialEq + Eq + Si fn decode_from_hex_str(s: &str) -> Option; } +impl StorableObjectId for String { + fn encode_to_hex_str(&self) -> String { + self.clone() + } + + fn decode_from_hex_str(s: &str) -> Option { + Some(s.to_owned()) + } +} + pub(crate) trait StorableObjectUpdate { fn id(&self) -> SO::Id; } diff --git a/src/event.rs b/src/event.rs index 0a3569755..f62f75a6d 100644 --- a/src/event.rs +++ b/src/event.rs @@ -10,6 +10,7 @@ use core::task::{Poll, Waker}; use std::collections::VecDeque; use std::ops::Deref; use std::sync::{Arc, Mutex}; +use std::time::{SystemTime, UNIX_EPOCH}; use bitcoin::blockdata::locktime::absolute::LockTime; use bitcoin::secp256k1::PublicKey; @@ -34,7 +35,9 @@ use lightning::{impl_writeable_tlv_based, impl_writeable_tlv_based_enum}; use lightning_liquidity::lsps2::utils::compute_opening_fee; use lightning_types::payment::{PaymentHash, PaymentPreimage}; -use crate::config::{may_announce_channel, Config, PEER_RECONNECTION_INTERVAL}; +use crate::config::{ + may_announce_channel, Config, ForwardedPaymentTrackingMode, PEER_RECONNECTION_INTERVAL, +}; use crate::connection::ConnectionManager; use crate::data_store::DataStoreUpdateResult; use crate::fee_estimator::ConfirmationTarget; @@ -51,11 +54,12 @@ use crate::payment::asynchronous::static_invoice_store::StaticInvoiceStore; use crate::payment::store::{ PaymentDetails, PaymentDetailsUpdate, PaymentDirection, PaymentKind, PaymentStatus, }; -use crate::payment::PaymentMetadata; +use crate::payment::{ChannelForwardingStats, ForwardedPaymentDetails, PaymentMetadata}; use crate::probing::Prober; use crate::runtime::Runtime; use crate::types::{ - CustomTlvRecord, DynStore, KeysManager, OnionMessenger, PaymentStore, Sweeper, Wallet, + ChannelForwardingStatsStore, CustomTlvRecord, DynStore, ForwardedPaymentStore, KeysManager, + OnionMessenger, PaymentStore, Sweeper, Wallet, }; use crate::{ hex_utils, BumpTransactionEventHandler, ChannelManager, Error, Graph, PeerInfo, PeerStore, @@ -551,6 +555,8 @@ where network_graph: Arc, liquidity_source: Arc>>, payment_store: Arc, + forwarded_payment_store: Arc, + channel_forwarding_stats_store: Arc, peer_store: Arc>, keys_manager: Arc, static_invoice_store: Option, @@ -572,6 +578,8 @@ where channel_manager: Arc, connection_manager: Arc>, output_sweeper: Arc, network_graph: Arc, liquidity_source: Arc>>, payment_store: Arc, + forwarded_payment_store: Arc, + channel_forwarding_stats_store: Arc, peer_store: Arc>, keys_manager: Arc, static_invoice_store: Option, onion_messenger: Arc, om_mailbox: Option>, prober: Option>, @@ -587,6 +595,8 @@ where network_graph, liquidity_source, payment_store, + forwarded_payment_store, + channel_forwarding_stats_store, peer_store, keys_manager, static_invoice_store, @@ -1751,6 +1761,105 @@ where .await; } + if let ([prev_htlc], [next_htlc], outbound_amount_msat, Some(fee_earned_msat)) = ( + prev_htlcs.as_slice(), + next_htlcs.as_slice(), + outbound_amount_forwarded_msat, + total_fee_earned_msat, + ) { + 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 = outbound_amount_msat.saturating_add(fee_earned_msat); + + // Inbound and outbound stats are persisted separately. A partial persistence + // failure may replay this event and count an update that already succeeded again. + // Persist the outbound update first so a failure between the writes cannot + // overcount fee revenue. + self.channel_forwarding_stats_store + .insert_or_update(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: outbound_amount_msat, + total_fee_earned_msat: Some(0), + total_skimmed_fee_msat: 0, + onchain_claims_count: u64::from(claim_from_onchain_tx), + first_forwarded_at_timestamp: forwarded_at_timestamp, + last_forwarded_at_timestamp: forwarded_at_timestamp, + }) + .await + .map_err(|e| { + log_error!( + self.logger, + "Failed to update outbound channel forwarding stats: {e}" + ); + ReplayEvent() + })?; + + self.channel_forwarding_stats_store + .insert_or_update(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: skimmed_fee_msat.unwrap_or(0), + onchain_claims_count: 0, + first_forwarded_at_timestamp: forwarded_at_timestamp, + last_forwarded_at_timestamp: forwarded_at_timestamp, + }) + .await + .map_err(|e| { + log_error!( + self.logger, + "Failed to update inbound channel forwarding stats: {e}" + ); + ReplayEvent() + })?; + + if matches!( + self.config.forwarded_payment_tracking_mode, + ForwardedPaymentTrackingMode::Detailed + ) { + self.forwarded_payment_store + .insert(ForwardedPaymentDetails { + id: hex_utils::to_string( + &self.keys_manager.get_secure_random_bytes(), + ), + prev_channel_id: prev_htlc.channel_id, + next_channel_id: next_htlc.channel_id, + prev_user_channel_id: prev_htlc.user_channel_id.map(UserChannelId), + next_user_channel_id: next_htlc.user_channel_id.map(UserChannelId), + prev_node_id: prev_htlc.node_id, + next_node_id: next_htlc.node_id, + inbound_amount_forwarded_msat: Some(inbound_amount_msat), + total_fee_earned_msat: Some(fee_earned_msat), + skimmed_fee_msat, + claim_from_onchain_tx, + outbound_amount_forwarded_msat: Some(outbound_amount_msat), + forwarded_at_timestamp, + }) + .await + .map_err(|e| { + log_error!(self.logger, "Failed to store forwarded payment: {e}"); + ReplayEvent() + })?; + } + } else { + log_debug!( + self.logger, + "Skipping forwarding payment tracking for forward with {} inbound and {} outbound HTLC(s), because tracking requires a single HTLC in each direction and a known fee", + prev_htlcs.len(), + next_htlcs.len() + ); + } + let event = Event::PaymentForwarded { prev_htlcs: prev_htlcs.into_iter().map(HTLCLocator::from).collect(), next_htlcs: next_htlcs.into_iter().map(HTLCLocator::from).collect(), @@ -2237,6 +2346,15 @@ 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]), + 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 +2579,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/io/mod.rs b/src/io/mod.rs index c70c68d96..87a639e84 100644 --- a/src/io/mod.rs +++ b/src/io/mod.rs @@ -33,6 +33,20 @@ pub(crate) const PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE: &str = ""; pub(crate) const PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE: &str = "pending_payments"; pub(crate) const PENDING_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE: &str = ""; +/// The forwarded payment information will be persisted under this prefix. +pub(crate) const FORWARDED_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE: &str = "forwarded_payments"; +pub(crate) const FORWARDED_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE: &str = ""; + +/// The channel forwarding stats will be persisted under this prefix. +pub(crate) const CHANNEL_FORWARDING_STATS_PERSISTENCE_PRIMARY_NAMESPACE: &str = + "channel_forwarding_stats"; +pub(crate) const CHANNEL_FORWARDING_STATS_PERSISTENCE_SECONDARY_NAMESPACE: &str = ""; + +/// The channel pair forwarding stats will be persisted under this prefix. +pub(crate) const CHANNEL_PAIR_FORWARDING_STATS_PERSISTENCE_PRIMARY_NAMESPACE: &str = + "channel_pair_forwarding_stats"; +pub(crate) const CHANNEL_PAIR_FORWARDING_STATS_PERSISTENCE_SECONDARY_NAMESPACE: &str = ""; + /// The 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 2bee539f7..064aafd18 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -171,9 +171,13 @@ use lnurl_auth::LnurlAuth; use logger::{log_debug, log_error, log_info, log_trace, LdkLogger, Logger}; use payment::asynchronous::om_mailbox::OnionMessageMailbox; use payment::asynchronous::static_invoice_store::StaticInvoiceStore; +use payment::forwarding_store::{ + aggregate_channel_pair_stats as aggregate_channel_pair_stats_impl, + run_forwarded_payment_aggregation, +}; use payment::{ - Bolt11Payment, Bolt12Payment, OnchainPayment, PaymentDetails, PaymentDetailsPage, - SpontaneousPayment, UnifiedPayment, + Bolt11Payment, Bolt12Payment, ChannelPairForwardingStats, Forwarding, OnchainPayment, + PaymentDetails, PaymentDetailsPage, SpontaneousPayment, UnifiedPayment, }; use peer_store::{PeerInfo, PeerStore}; #[cfg(feature = "uniffi")] @@ -182,7 +186,8 @@ use probing::{run_prober, Prober}; use runtime::Runtime; pub use tokio; use types::{ - Broadcaster, BumpTransactionEventHandler, ChainMonitor, ChannelManager, DynStore, Graph, + Broadcaster, BumpTransactionEventHandler, ChainMonitor, ChannelForwardingStatsStore, + ChannelManager, ChannelPairForwardingStatsStore, DynStore, ForwardedPaymentStore, Graph, HRNResolver, KeysManager, OnionMessenger, PaymentStore, PeerManager, Router, Scorer, Sweeper, Wallet, }; @@ -265,6 +270,10 @@ pub struct Node { scorer: Arc>, peer_store: Arc>>, payment_store: Arc, + forwarded_payment_store: Arc, + channel_forwarding_stats_store: Arc, + channel_pair_forwarding_stats_store: Arc, + forwarded_payment_aggregation_retention_secs: u64, lnurl_auth: Arc, is_running: Arc>, node_metrics: Arc, @@ -647,6 +656,22 @@ 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 forwarded_payment_store = Arc::clone(&self.forwarded_payment_store); + let channel_pair_stats_store = Arc::clone(&self.channel_pair_forwarding_stats_store); + let aggregation_logger = Arc::clone(&self.logger); + self.runtime.spawn_cancellable_background_task(async move { + run_forwarded_payment_aggregation( + stop_aggregation, + forwarded_payment_store, + channel_pair_stats_store, + retention_secs, + aggregation_logger, + ) + .await; + }); + let bump_tx_event_handler = Arc::new(BumpTransactionEventHandler::new( Arc::clone(&self.tx_broadcaster), Arc::new(LdkWallet::new(Arc::clone(&self.wallet), Arc::clone(&self.logger))), @@ -671,6 +696,8 @@ impl Node { Arc::clone(&self.network_graph), Arc::clone(&self.liquidity_source), Arc::clone(&self.payment_store), + Arc::clone(&self.forwarded_payment_store), + Arc::clone(&self.channel_forwarding_stats_store), Arc::clone(&self.peer_store), Arc::clone(&self.keys_manager), static_invoice_store, @@ -1184,6 +1211,30 @@ impl Node { )) } + /// Returns a handler allowing to query forwarded payments and forwarding statistics. + #[cfg(not(feature = "uniffi"))] + pub fn forwarding(&self) -> Forwarding { + Forwarding::new( + Arc::clone(&self.runtime), + Arc::clone(&self.forwarded_payment_store), + Arc::clone(&self.channel_forwarding_stats_store), + Arc::clone(&self.channel_pair_forwarding_stats_store), + Arc::clone(&self.config), + ) + } + + /// Returns a handler allowing to query forwarded payments and forwarding statistics. + #[cfg(feature = "uniffi")] + pub fn forwarding(&self) -> Arc { + Arc::new(Forwarding::new( + Arc::clone(&self.runtime), + Arc::clone(&self.forwarded_payment_store), + Arc::clone(&self.channel_forwarding_stats_store), + Arc::clone(&self.channel_pair_forwarding_stats_store), + Arc::clone(&self.config), + )) + } + /// Authenticates the user via [LNURL-auth] for the given LNURL string. /// /// [LNURL-auth]: https://github.com/lnurl/luds/blob/luds/04.md @@ -2546,6 +2597,32 @@ async fn connect_and_discover_lsp( } } +/// 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`]: ChannelPairForwardingStats::bucket_start_timestamp +/// [`bucket_size_secs`]: ChannelPairForwardingStats::bucket_size_secs +/// [`id`]: ChannelPairForwardingStats::id +/// [`aggregated_at_timestamp`]: ChannelPairForwardingStats::aggregated_at_timestamp +pub fn aggregate_channel_pair_stats( + buckets: &[ChannelPairForwardingStats], +) -> Option { + aggregate_channel_pair_stats_impl(buckets) +} + #[cfg(test)] mod tests { use lightning::util::ser::{Readable, Writeable}; diff --git a/src/payment/forwarding.rs b/src/payment/forwarding.rs new file mode 100644 index 000000000..cf8cc0c3d --- /dev/null +++ b/src/payment/forwarding.rs @@ -0,0 +1,403 @@ +// This file is Copyright its original authors, visible in version control history. +// +// This file is licensed under the Apache License, Version 2.0 or the MIT license , at your option. You may not use this file except in +// accordance with one or both of these licenses. + +//! Holds a handler allowing to query forwarded payments and forwarding statistics. + +use std::sync::Arc; + +use bitcoin::secp256k1::PublicKey; +use lightning::impl_writeable_tlv_based; +use lightning::ln::types::ChannelId; + +use crate::config::{Config, ForwardedPaymentTrackingMode}; +use crate::error::Error; +use crate::ffi::{maybe_deref, maybe_wrap}; +use crate::runtime::Runtime; +use crate::types::{ + ChannelForwardingStatsStore, ChannelPairForwardingStatsStore, ForwardedPaymentStore, +}; +use crate::UserChannelId; + +use super::PageToken; + +/// 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, randomly generated identifier for this forwarded payment. + pub id: String, + /// 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: String, + /// 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`]. +/// +/// [`Node::forwarding`]: crate::Node::forwarding +#[cfg_attr(feature = "uniffi", derive(uniffi::Object))] +pub struct Forwarding { + runtime: Arc, + forwarded_payment_store: Arc, + channel_forwarding_stats_store: Arc, + channel_pair_forwarding_stats_store: Arc, + config: Arc, +} + +impl Forwarding { + pub(crate) fn new( + runtime: Arc, forwarded_payment_store: Arc, + channel_forwarding_stats_store: Arc, + channel_pair_forwarding_stats_store: Arc, + config: Arc, + ) -> Self { + Self { + runtime, + forwarded_payment_store, + channel_forwarding_stats_store, + channel_pair_forwarding_stats_store, + config, + } + } + + /// Retrieves all forwarded payments that match the given predicate. + pub fn list_payments_with_filter bool>( + &self, f: F, + ) -> Result, Error> { + self.runtime.block_on(self.forwarded_payment_store.list_filter(f)) + } + + /// Retrieves all channel forwarding statistics that match the given predicate. + pub fn list_channel_stats_with_filter bool>( + &self, f: F, + ) -> Vec { + self.runtime.block_on(self.channel_forwarding_stats_store.list_filter(f)) + } + + /// Retrieves all channel pair forwarding statistics that match the given predicate. + pub fn list_channel_pair_stats_with_filter bool>( + &self, f: F, + ) -> Result, Error> { + self.runtime.block_on(self.channel_pair_forwarding_stats_store.list_filter(f)) + } + + /// Retrieves pages of channel-pair forwarding statistics until the given filter produces a + /// non-empty result page or the underlying store is exhausted. + fn list_channel_pair_stats_filtered_page bool>( + &self, mut page_token: Option, mut f: F, + ) -> Result { + loop { + let ldk_page_token = page_token.as_ref().map(|token| maybe_deref(token).clone()); + let (mut stats, next_page_token) = self + .runtime + .block_on(self.channel_pair_forwarding_stats_store.list_page(ldk_page_token))?; + let next_page_token = next_page_token.map(maybe_wrap); + stats.retain(|stats| f(stats)); + if !stats.is_empty() || next_page_token.is_none() { + stats.sort_by_key(|stats| stats.bucket_start_timestamp); + return Ok(ChannelPairForwardingStatsPage { stats, next_page_token }); + } + page_token = next_page_token; + } + } +} + +#[cfg_attr(feature = "uniffi", uniffi::export)] +impl Forwarding { + /// Returns the configured forwarded payment tracking mode. + pub fn tracking_mode(&self) -> ForwardedPaymentTrackingMode { + self.config.forwarded_payment_tracking_mode + } + + /// Retrieve the details of a specific forwarded payment using its opaque identifier. + /// + /// The identifier is returned in [`ForwardedPaymentDetails::id`]. + pub fn payment( + &self, forwarded_payment_id: String, + ) -> Result, Error> { + self.runtime.block_on(self.forwarded_payment_store.get(&forwarded_payment_id)) + } + + /// Retrieves a page of forwarded payments from the underlying paginated store. + pub fn list_payments( + &self, page_token: Option, + ) -> Result { + let ldk_page_token = page_token.as_ref().map(|token| maybe_deref(token).clone()); + let (payments, next_page_token) = + self.runtime.block_on(self.forwarded_payment_store.list_page(ldk_page_token))?; + let next_page_token = next_page_token.map(maybe_wrap); + Ok(ForwardedPaymentDetailsPage { payments, next_page_token }) + } + + /// Retrieve the forwarding statistics for a specific channel. + pub fn channel_stats( + &self, channel_id: &ChannelId, + ) -> Result, Error> { + self.runtime.block_on(self.channel_forwarding_stats_store.get(channel_id)) + } + + /// Retrieves a page of channel forwarding statistics from the underlying paginated store. + pub fn list_channel_stats( + &self, page_token: Option, + ) -> Result { + let ldk_page_token = page_token.as_ref().map(|token| maybe_deref(token).clone()); + let page = + self.runtime.block_on(self.channel_forwarding_stats_store.list_page(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 (stats, next_page_token) = self + .runtime + .block_on(self.channel_pair_forwarding_stats_store.list_page(ldk_page_token))?; + let next_page_token = next_page_token.map(maybe_wrap); + Ok(ChannelPairForwardingStatsPage { stats, next_page_token }) + } + + /// Retrieves a page of channel pair forwarding statistics within a specific time range. + /// + /// A bucket matches when its start timestamp is greater than or equal to `start_timestamp` and + /// less than `end_timestamp`. + /// + /// [`PaginatedKVStore`] lists newest-written first and gives no way to seek to a key or a time, + /// so a range near the present is reached early and an older one only after paging back past + /// everything written since. Filtering happens here rather than in the store, so a range that + /// matches nothing costs a walk over every bucket. A backend that departs from that ordering + /// only changes which page a match lands on, never whether it is found. + /// + /// [`PaginatedKVStore`]: lightning::util::persist::PaginatedKVStore + /// + /// The listing is complete when `next_page_token` is `None`. Matches within each returned page + /// are sorted by bucket start time, but ordering is not global across pages. + pub fn list_channel_pair_stats_in_range( + &self, start_timestamp: u64, end_timestamp: u64, page_token: Option, + ) -> Result { + self.list_channel_pair_stats_filtered_page(page_token, |stats| { + stats.bucket_start_timestamp >= start_timestamp + && stats.bucket_start_timestamp < end_timestamp + }) + } + + /// Retrieves a page of forwarding statistics buckets for a specific channel pair. + /// + /// Buckets for a pair are not stored together, so filtering happens here rather than in the + /// store, and paging continues until a page yields a match. A pair the node writes often is + /// therefore found within the first page or two, while a long-idle pair, or one that has never + /// forwarded, costs a walk over every bucket written since it was last active. + /// + /// The listing is complete when `next_page_token` is `None`. Matches within each returned page + /// are sorted by bucket start time, but ordering is not global across pages. + pub fn list_channel_pair_stats_for_pair( + &self, prev_channel_id: ChannelId, next_channel_id: ChannelId, + page_token: Option, + ) -> Result { + self.list_channel_pair_stats_filtered_page(page_token, |stats| { + stats.prev_channel_id == prev_channel_id && stats.next_channel_id == next_channel_id + }) + } +} diff --git a/src/payment/forwarding_store.rs b/src/payment/forwarding_store.rs new file mode 100644 index 000000000..4e7ed6f05 --- /dev/null +++ b/src/payment/forwarding_store.rs @@ -0,0 +1,1190 @@ +// 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; +use std::marker::PhantomData; +use std::sync::Arc; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use bitcoin::hashes::{sha256, Hash}; +use bitcoin::secp256k1::PublicKey; +use lightning::ln::types::ChannelId; +use lightning::util::logger::Logger as _; +use lightning::util::persist::{KVStore, PageToken, PaginatedKVStore}; + +use super::forwarding::{ + ChannelForwardingStats, ChannelPairForwardingStats, ForwardedPaymentDetails, +}; +use crate::data_store::{StorableObject, StorableObjectId, StorableObjectUpdate, UpdatableObject}; +use crate::hex_utils; +use crate::logger::{log_debug, log_error, Logger}; +use crate::types::{DynStore, DynStoreRef}; +use crate::Error; + +pub(crate) const FORWARDED_PAYMENT_AGGREGATION_BUCKET_SIZE_SECS: u64 = 60 * 60; + +/// A disk-backed store for forwarding data that is too large to keep in memory. +pub(crate) struct DiskStore { + mutation_lock: tokio::sync::Mutex<()>, + primary_namespace: String, + secondary_namespace: String, + kv_store: Arc, + logger: Arc, + _object: PhantomData, +} + +impl DiskStore { + pub(crate) fn new( + primary_namespace: String, secondary_namespace: String, kv_store: Arc, + logger: Arc, + ) -> Self { + Self { + mutation_lock: tokio::sync::Mutex::new(()), + primary_namespace, + secondary_namespace, + kv_store, + logger, + _object: PhantomData, + } + } + + pub(crate) async fn insert(&self, object: SO) -> Result<(), Error> { + let _guard = self.mutation_lock.lock().await; + self.persist_unlocked(&object).await + } + + pub(crate) async fn get(&self, id: &SO::Id) -> Result, Error> { + let _guard = self.mutation_lock.lock().await; + self.get_unlocked(id).await + } + + pub(crate) async fn contains_key(&self, id: &SO::Id) -> Result { + Ok(self.get(id).await?.is_some()) + } + + pub(crate) async fn remove(&self, id: &SO::Id) -> Result<(), Error> { + let _guard = self.mutation_lock.lock().await; + let store_key = id.encode_to_hex_str(); + KVStore::remove( + &*self.kv_store, + &self.primary_namespace, + &self.secondary_namespace, + &store_key, + false, + ) + .await + .map_err(|e| { + log_error!( + self.logger, + "Removing object data for key {}/{}/{} failed due to: {}", + &self.primary_namespace, + &self.secondary_namespace, + store_key, + e + ); + Error::PersistenceFailed + }) + } + + pub(crate) async fn is_empty(&self) -> Result { + let _guard = self.mutation_lock.lock().await; + let response = self.list_keys_page_unlocked(None).await?; + Ok(response.keys.is_empty()) + } + + /// Returns every object matching `f`. + /// + /// The mutation lock is taken per page rather than for the whole walk, so this is not a + /// snapshot: an object written while the walk is in progress may or may not be included. Taking + /// it once for the whole walk would block writers for as long as the store takes to read, which + /// on a remote [`KVStore`] is one round trip per object. + pub(crate) async fn list_filter bool>( + &self, mut f: F, + ) -> Result, Error> { + let mut objects = Vec::new(); + let mut page_token = None; + loop { + let (page, next_page_token) = self.list_page(page_token).await?; + objects.extend(page.iter().filter(|object| f(object)).cloned()); + let Some(next_page_token) = next_page_token else { + return Ok(objects); + }; + page_token = Some(next_page_token); + } + } + + pub(crate) async fn list_page( + &self, page_token: Option, + ) -> Result<(Vec, Option), Error> { + let _guard = self.mutation_lock.lock().await; + self.list_page_unlocked(page_token).await + } + + async fn get_unlocked(&self, id: &SO::Id) -> Result, Error> { + let store_key = id.encode_to_hex_str(); + let data = match KVStore::read( + &DynStoreRef(Arc::clone(&self.kv_store)), + &self.primary_namespace, + &self.secondary_namespace, + &store_key, + ) + .await + { + Ok(data) => data, + Err(e) if e.kind() == bitcoin::io::ErrorKind::NotFound => return Ok(None), + Err(e) => { + log_error!( + self.logger, + "Reading object data for key {}/{}/{} failed due to: {}", + &self.primary_namespace, + &self.secondary_namespace, + store_key, + e + ); + return Err(Error::PersistenceFailed); + }, + }; + + SO::read(&mut &data[..]).map(Some).map_err(|e| { + log_error!( + self.logger, + "Failed to deserialize object data for key {}/{}/{}: {}", + &self.primary_namespace, + &self.secondary_namespace, + store_key, + e + ); + Error::PersistenceFailed + }) + } + + async fn list_page_unlocked( + &self, page_token: Option, + ) -> Result<(Vec, Option), Error> { + let response = self.list_keys_page_unlocked(page_token).await?; + let mut objects = Vec::with_capacity(response.keys.len()); + for key in response.keys { + let object = self.get_unlocked_key(&key).await?; + objects.push(object); + } + Ok((objects, response.next_page_token)) + } + + async fn list_keys_page_unlocked( + &self, page_token: Option, + ) -> Result { + PaginatedKVStore::list_paginated( + &DynStoreRef(Arc::clone(&self.kv_store)), + &self.primary_namespace, + &self.secondary_namespace, + page_token, + ) + .await + .map_err(|e| { + log_error!( + self.logger, + "Listing object data under {}/{} failed due to: {}", + &self.primary_namespace, + &self.secondary_namespace, + e + ); + Error::PersistenceFailed + }) + } + + async fn get_unlocked_key(&self, key: &str) -> Result { + let data = KVStore::read( + &DynStoreRef(Arc::clone(&self.kv_store)), + &self.primary_namespace, + &self.secondary_namespace, + key, + ) + .await + .map_err(|e| { + log_error!( + self.logger, + "Reading object data for key {}/{}/{} failed due to: {}", + &self.primary_namespace, + &self.secondary_namespace, + key, + e + ); + Error::PersistenceFailed + })?; + + SO::read(&mut &data[..]).map_err(|e| { + log_error!( + self.logger, + "Failed to deserialize object data for key {}/{}/{}: {}", + &self.primary_namespace, + &self.secondary_namespace, + key, + e + ); + Error::PersistenceFailed + }) + } + + async fn persist_unlocked(&self, object: &SO) -> Result<(), Error> { + let store_key = object.id().encode_to_hex_str(); + KVStore::write( + &*self.kv_store, + &self.primary_namespace, + &self.secondary_namespace, + &store_key, + object.encode(), + ) + .await + .map_err(|e| { + log_error!( + self.logger, + "Write for key {}/{}/{} failed due to: {}", + &self.primary_namespace, + &self.secondary_namespace, + store_key, + e + ); + Error::PersistenceFailed + }) + } +} + +impl StorableObject for ForwardedPaymentDetails { + type Id = String; + + fn id(&self) -> Self::Id { + self.id.clone() + } +} + +fn channel_pair_stats_id( + prev: &ChannelId, next: &ChannelId, bucket_start_timestamp: u64, +) -> String { + // 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()); + sha256::Hash::hash(&bytes).to_string() +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) struct ChannelForwardingStatsUpdate { + pub channel_id: ChannelId, + pub counterparty_node_id: Option, + pub inbound_payments_increment: u64, + pub outbound_payments_increment: u64, + pub inbound_amount_increment_msat: u64, + pub outbound_amount_increment_msat: u64, + pub fee_earned_increment_msat: Option, + pub skimmed_fee_increment_msat: u64, + pub onchain_claims_increment: u64, + pub timestamp: u64, +} + +impl StorableObjectUpdate for ChannelForwardingStatsUpdate { + fn id(&self) -> ChannelId { + self.channel_id + } +} + +impl StorableObjectId for ChannelId { + fn encode_to_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 ChannelForwardingStats { + type Id = ChannelId; + + fn id(&self) -> Self::Id { + self.channel_id + } +} + +impl UpdatableObject for ChannelForwardingStats { + 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; + } + if update.inbound_payments_increment > 0 { + self.inbound_payments_forwarded += update.inbound_payments_increment; + updated = true; + } + if update.outbound_payments_increment > 0 { + self.outbound_payments_forwarded += update.outbound_payments_increment; + updated = true; + } + if update.inbound_amount_increment_msat > 0 { + self.total_inbound_amount_msat += update.inbound_amount_increment_msat; + updated = true; + } + if update.outbound_amount_increment_msat > 0 { + self.total_outbound_amount_msat += update.outbound_amount_increment_msat; + updated = true; + } + match (self.total_fee_earned_msat.as_mut(), update.fee_earned_increment_msat) { + (Some(total), Some(increment)) if increment > 0 => { + *total += increment; + updated = true; + }, + (Some(_), None) => { + self.total_fee_earned_msat = None; + updated = true; + }, + _ => {}, + } + if update.skimmed_fee_increment_msat > 0 { + self.total_skimmed_fee_msat += update.skimmed_fee_increment_msat; + updated = true; + } + if update.onchain_claims_increment > 0 { + self.onchain_claims_count += update.onchain_claims_increment; + updated = true; + } + if updated { + self.first_forwarded_at_timestamp = + self.first_forwarded_at_timestamp.min(update.timestamp); + self.last_forwarded_at_timestamp = + self.last_forwarded_at_timestamp.max(update.timestamp); + } + updated + } + + fn to_update(&self) -> Self::Update { + ChannelForwardingStatsUpdate { + channel_id: self.channel_id, + counterparty_node_id: self.counterparty_node_id, + inbound_payments_increment: self.inbound_payments_forwarded, + outbound_payments_increment: self.outbound_payments_forwarded, + inbound_amount_increment_msat: self.total_inbound_amount_msat, + outbound_amount_increment_msat: self.total_outbound_amount_msat, + fee_earned_increment_msat: self.total_fee_earned_msat, + skimmed_fee_increment_msat: self.total_skimmed_fee_msat, + onchain_claims_increment: self.onchain_claims_count, + timestamp: self.last_forwarded_at_timestamp, + } + } +} + +impl StorableObject for ChannelPairForwardingStats { + type Id = String; + + fn id(&self) -> Self::Id { + self.id.clone() + } +} + +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( + forwarded_payment_store: &DiskStore, + channel_pair_stats_store: &DiskStore, retention_secs: u64, + logger: &Arc, +) { + match aggregate_expired_forwarded_payments( + forwarded_payment_store, + channel_pair_stats_store, + retention_secs, + logger, + ) + .await + { + Ok((pair_count, payment_count)) if pair_count > 0 => { + log_debug!( + logger, + "Aggregated {} forwarded payments into {} channel pair buckets", + payment_count, + pair_count + ); + }, + Ok((0, payment_count)) if payment_count > 0 => { + log_debug!( + logger, + "Removed {} forwarded payment details from previously aggregated buckets", + payment_count + ); + }, + Err(e) => log_error!(logger, "Forwarded payment aggregation failed: {}", e), + _ => {}, + } +} + +pub(crate) async fn run_forwarded_payment_aggregation( + mut stop_receiver: tokio::sync::watch::Receiver<()>, + forwarded_payment_store: Arc>, + channel_pair_stats_store: Arc>, retention_secs: u64, + logger: Arc, +) { + if retention_secs == 0 { + match forwarded_payment_store.is_empty().await { + Ok(true) => return, + Ok(false) => {}, + Err(e) => log_error!(logger, "Failed to check forwarded payment store: {}", e), + } + } + + aggregate_forwarded_payments_and_log( + &forwarded_payment_store, + &channel_pair_stats_store, + retention_secs, + &logger, + ) + .await; + + if retention_secs == 0 { + match forwarded_payment_store.is_empty().await { + Ok(true) => return, + Ok(false) => {}, + Err(e) => log_error!(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( + &forwarded_payment_store, + &channel_pair_stats_store, + retention_secs, + &logger, + ) + .await; + if retention_secs == 0 { + match forwarded_payment_store.is_empty().await { + Ok(true) => break, + Ok(false) => {}, + Err(e) => log_error!(logger, "Failed to check forwarded payment store: {}", e), + } + } + } + } + } +} + +/// Aggregate forwarded payments older than the configured retention period into fixed-width +/// channel-pair statistics buckets. +pub(crate) async fn aggregate_expired_forwarded_payments( + forwarded_payment_store: &DiskStore, + channel_pair_stats_store: &DiskStore, 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, + 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: &DiskStore, + channel_pair_stats_store: &DiskStore, 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 page_token = None; + loop { + let (page, next_page_token) = forwarded_payment_store.list_page(page_token).await?; + for payment in page { + if payment.forwarded_at_timestamp >= oldest_retained_bucket_start { + continue; + } + let bucket_start = + (payment.forwarded_at_timestamp / bucket_size_secs) * bucket_size_secs; + bucket_groups + .entry((payment.prev_channel_id, payment.next_channel_id, bucket_start)) + .or_default() + .push(payment); + } + let Some(next_page_token) = 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.contains_key(&pair_id).await? { + 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.clone(), + 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. +/// +/// Returns `None` if `buckets` is empty or contains statistics for different channel pairs. See the +/// public wrapper, [`crate::aggregate_channel_pair_stats`], for how the bucket-shaped fields of the +/// result read. +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; + + use super::*; + use crate::io::sqlite_store::SqliteStore; + use crate::io::test_utils::{random_storage_path, InMemoryStore}; + use crate::types::{DynStore, DynStoreWrapper}; + + type TestForwardedPaymentStore = DiskStore; + type TestChannelPairStatsStore = DiskStore; + + fn test_stores() -> (TestForwardedPaymentStore, TestChannelPairStatsStore, Arc) { + let kv_store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let logger = Arc::new(Logger::new_log_facade()); + let forwarded_payment_store = DiskStore::new( + "test_forwarded_payments".to_string(), + String::new(), + Arc::clone(&kv_store), + Arc::clone(&logger), + ); + let channel_pair_stats_store = DiskStore::new( + "test_channel_pair_stats".to_string(), + String::new(), + kv_store, + Arc::clone(&logger), + ); + (forwarded_payment_store, channel_pair_stats_store, logger) + } + + fn forwarded_payment( + id: u8, forwarded_at_timestamp: u64, inbound_amount_msat: u64, outbound_amount_msat: u64, + fee_msat: u64, + ) -> ForwardedPaymentDetails { + ForwardedPaymentDetails { + id: hex_utils::to_string(&[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()); + } + + #[tokio::test] + async fn aggregation_retains_current_and_previous_buckets() { + let (forwarded_payment_store, channel_pair_stats_store, logger) = test_stores(); + let closed_bucket_payment = forwarded_payment(1, 899, 110, 100, 10); + let partial_bucket_payment = forwarded_payment(2, 939, 220, 200, 20); + forwarded_payment_store.insert(closed_bucket_payment.clone()).await.unwrap(); + forwarded_payment_store.insert(partial_bucket_payment.clone()).await.unwrap(); + + // At timestamp 1,000, the current bucket starts at 960 and the previous bucket starts at + // 900. Only payments older than the previous bucket are aggregated. + assert_eq!( + aggregate_expired_forwarded_payments_at( + &forwarded_payment_store, + &channel_pair_stats_store, + 60, + 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 background_aggregation_runs_immediately() { + let (forwarded_payment_store, channel_pair_stats_store, logger) = test_stores(); + let forwarded_payment_store = Arc::new(forwarded_payment_store); + let channel_pair_stats_store = Arc::new(channel_pair_stats_store); + let payment = forwarded_payment(1, 1, 110, 100, 10); + forwarded_payment_store.insert(payment.clone()).await.unwrap(); + let (_stop_sender, stop_receiver) = tokio::sync::watch::channel(()); + + run_forwarded_payment_aggregation( + stop_receiver, + Arc::clone(&forwarded_payment_store), + Arc::clone(&channel_pair_stats_store), + 0, + logger, + ) + .await; + + assert!(forwarded_payment_store.is_empty().await.unwrap()); + let bucket_id = + channel_pair_stats_id(&payment.prev_channel_id, &payment.next_channel_id, 0); + assert_eq!( + channel_pair_stats_store.get(&bucket_id).await.unwrap().unwrap().payment_count, + 1 + ); + } + + #[tokio::test] + async fn aggregation_preserves_unknown_fees() { + let (forwarded_payment_store, channel_pair_stats_store, logger) = test_stores(); + let known_fee_payment = forwarded_payment(1, 850, 110, 100, 10); + let mut unknown_fee_payment = forwarded_payment(2, 851, 200, 200, 0); + unknown_fee_payment.total_fee_earned_msat = None; + forwarded_payment_store.insert(known_fee_payment).await.unwrap(); + forwarded_payment_store.insert(unknown_fee_payment).await.unwrap(); + + assert_eq!( + aggregate_expired_forwarded_payments_at( + &forwarded_payment_store, + &channel_pair_stats_store, + 60, + 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 = ChannelForwardingStats { + channel_id: ChannelId([1; 32]), + counterparty_node_id: None, + inbound_payments_forwarded: 1, + outbound_payments_forwarded: 0, + total_inbound_amount_msat: 110, + total_outbound_amount_msat: 0, + total_fee_earned_msat: Some(10), + total_skimmed_fee_msat: 0, + onchain_claims_count: 0, + first_forwarded_at_timestamp: 850, + last_forwarded_at_timestamp: 850, + }; + + assert!(stats.update(ChannelForwardingStatsUpdate { + channel_id: stats.channel_id, + counterparty_node_id: None, + inbound_payments_increment: 1, + outbound_payments_increment: 0, + inbound_amount_increment_msat: 200, + outbound_amount_increment_msat: 0, + fee_earned_increment_msat: None, + skimmed_fee_increment_msat: 0, + onchain_claims_increment: 0, + timestamp: 851, + })); + assert_eq!(stats.total_fee_earned_msat, None); + } + + #[tokio::test] + async fn aggregation_keeps_both_retained_bucket_boundaries() { + let (forwarded_payment_store, channel_pair_stats_store, logger) = test_stores(); + let older_bucket_payment = forwarded_payment(1, 839, 110, 100, 10); + let previous_bucket_payment = forwarded_payment(2, 840, 220, 200, 20); + let current_bucket_payment = forwarded_payment(3, 900, 330, 300, 30); + forwarded_payment_store.insert(older_bucket_payment.clone()).await.unwrap(); + forwarded_payment_store.insert(previous_bucket_payment.clone()).await.unwrap(); + forwarded_payment_store.insert(current_bucket_payment.clone()).await.unwrap(); + + assert_eq!( + aggregate_expired_forwarded_payments_at( + &forwarded_payment_store, + &channel_pair_stats_store, + 60, + 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, channel_pair_stats_store, logger) = test_stores(); + let payment = forwarded_payment(1, 899, 110, 100, 10); + forwarded_payment_store.insert(payment.clone()).await.unwrap(); + + assert_eq!( + aggregate_expired_forwarded_payments_at( + &forwarded_payment_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, + &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, channel_pair_stats_store, logger) = test_stores(); + let remaining_payment = forwarded_payment(2, 851, 220, 200, 20); + let committed_stats = channel_pair_stats(840, 60, 950, 2, 330, 300, 30, 850, 851); + channel_pair_stats_store.insert(committed_stats.clone()).await.unwrap(); + forwarded_payment_store.insert(remaining_payment.clone()).await.unwrap(); + + // This represents a retry after the bucket write and one of two detail deletions + // succeeded. The existing bucket is the commit marker, so its totals must not change. + assert_eq!( + aggregate_expired_forwarded_payments_at( + &forwarded_payment_store, + &channel_pair_stats_store, + 60, + 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) + ); + } + + #[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, channel_pair_stats_store, logger) = test_stores(); + let old_stats = channel_pair_stats(0, 3_600, 7_200, 1, 110, 100, 10, 100, 100); + let payment = forwarded_payment(2, 4_000, 220, 200, 20); + channel_pair_stats_store.insert(old_stats.clone()).await.unwrap(); + forwarded_payment_store.insert(payment.clone()).await.unwrap(); + + assert_eq!( + aggregate_expired_forwarded_payments_at( + &forwarded_payment_store, + &channel_pair_stats_store, + 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] + async fn channel_pair_stats_persist_to_sqlite() { + let mut storage_path = random_storage_path(); + storage_path.push("channel_pair_stats_persist_to_sqlite"); + let sqlite_store = + SqliteStore::new(storage_path, Some("stats.sqlite".to_string()), None).unwrap(); + let kv_store: Arc = Arc::new(DynStoreWrapper(sqlite_store)); + let logger = Arc::new(Logger::new_log_facade()); + let namespace = "sqlite_channel_pair_stats"; + let stats_store = + DiskStore::new(namespace.to_string(), String::new(), Arc::clone(&kv_store), logger); + let stats = channel_pair_stats(840, 60, 1_000, 2, 330, 300, 30, 850, 851); + + stats_store.insert(stats.clone()).await.unwrap(); + let keys = KVStore::list(&*kv_store, namespace, "").await.unwrap(); + assert_eq!(keys, vec![stats.id.encode_to_hex_str()]); + let bytes = KVStore::read(&*kv_store, namespace, "", &keys[0]).await.unwrap(); + assert_eq!(ChannelPairForwardingStats::read(&mut &bytes[..]).unwrap(), stats); + } + + #[tokio::test] + async fn disk_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 = DiskStore::new( + namespace.to_string(), + String::new(), + 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(); + } + + // Recreate the store to verify it has no in-memory state to preload. + let reopened_store = DiskStore::new(namespace.to_string(), String::new(), 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()) + ); + + let (first_page, next_page_token) = reopened_store.list_page(None).await.unwrap(); + assert_eq!(first_page.len(), 50); + let (second_page, next_page_token) = + reopened_store.list_page(next_page_token).await.unwrap(); + assert_eq!(second_page, vec![oldest_payment.clone()]); + assert!(next_page_token.is_none()); + + assert_eq!( + reopened_store.list_filter(|payment| payment.id == oldest_payment.id).await.unwrap(), + vec![oldest_payment] + ); + } +} diff --git a/src/payment/mod.rs b/src/payment/mod.rs index b0f4901a7..1aa95ead8 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; mod onchain; pub(crate) mod pending_payment_store; mod spontaneous; @@ -19,6 +21,11 @@ mod unified; pub use bolt11::Bolt11Payment; pub(crate) use bolt11::PaymentMetadata; pub use bolt12::{Bolt12Payment, PayerProofOptions}; +pub use forwarding::{ + ChannelForwardingStats, ChannelForwardingStatsPage, ChannelPairForwardingStats, + ChannelPairForwardingStatsPage, ForwardedPaymentDetails, ForwardedPaymentDetailsPage, + Forwarding, +}; pub use onchain::OnchainPayment; pub(crate) use pending_payment_store::{FundingTxCandidate, PendingPaymentDetails}; pub use spontaneous::SpontaneousPayment; diff --git a/src/types.rs b/src/types.rs index 65156982e..6c2981223 100644 --- a/src/types.rs +++ b/src/types.rs @@ -50,7 +50,11 @@ 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::forwarding_store::DiskStore; +use crate::payment::{ + ChannelForwardingStats, ChannelPairForwardingStats, ForwardedPaymentDetails, PaymentDetails, + PendingPaymentDetails, +}; use crate::runtime::RuntimeSpawner; #[cfg(feature = "uniffi")] @@ -376,6 +380,9 @@ pub(crate) type BumpTransactionEventHandler = >; pub(crate) type PaymentStore = DataStore, KeepLeastRecentlyUsed>; +pub(crate) type ForwardedPaymentStore = DiskStore; +pub(crate) type ChannelForwardingStatsStore = DataStore>; +pub(crate) type ChannelPairForwardingStatsStore = DiskStore; /// A local, potentially user-provided, identifier of a channel. /// diff --git a/tests/integration_tests_rust.rs b/tests/integration_tests_rust.rs index fd247f74c..8180774c1 100644 --- a/tests/integration_tests_rust.rs +++ b/tests/integration_tests_rust.rs @@ -35,7 +35,8 @@ 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; @@ -730,6 +731,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(); + 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().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(); + 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.clone()).unwrap().as_ref(), Some(details)); + assert!(forwarding.payment("not-a-known-id".to_string()).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)]