Add forwarded payment tracking and statistics

Routing nodes and LSPs want to track forwarded payments so they can run accounting on fees earned and track profitability across time. We now store these to make it easier to track and allows for future accounting utils in the future. To prevent potential privacy footguns, we only store the individual forwarding events for a given time period, and then we aggregate them into a single event for each channel pair, this prevents potential payment correlation.

For lightweight nodes we have the `Stats` mode that only tracks the total forwarding stats per channel.

Co-Authored-By: Claude Sonnet 4.5 (1M context) <noreply@anthropic.com>

Author: benthecarman

bindings/ldk_node.udl

@@ -13,6 +13,7 @@ dictionary Config {
 	u64 probing_liquidity_limit_multiplier;
 	AnchorChannelsConfig? anchor_channels_config;
 	RouteParametersConfig? route_parameters;
+	ForwardedPaymentTrackingMode forwarded_payment_tracking_mode;
 };
 
 dictionary AnchorChannelsConfig {
@@ -189,6 +190,12 @@ interface Node {
 	void remove_payment([ByRef]PaymentId payment_id);
 	BalanceDetails list_balances();
 	sequence<PaymentDetails> list_payments();
+	ForwardedPaymentDetails? forwarded_payment([ByRef]ForwardedPaymentId forwarded_payment_id);
+	sequence<ForwardedPaymentDetails> list_forwarded_payments();
+	ForwardedPaymentTrackingMode forwarded_payment_tracking_mode();
+	ChannelForwardingStats? channel_forwarding_stats([ByRef]ChannelId channel_id);
+	sequence<ChannelForwardingStats> list_channel_forwarding_stats();
+	sequence<ChannelPairForwardingStats> list_channel_pair_forwarding_stats();
 	sequence<PeerDetails> list_peers();
 	sequence<ChannelDetails> list_channels();
 	NetworkGraph network_graph();
@@ -486,6 +493,12 @@ enum PaymentStatus {
 	"Failed",
 };
 
+[Enum]
+interface ForwardedPaymentTrackingMode {
+	Detailed(u64 retention_minutes);
+	Stats();
+};
+
 dictionary LSPFeeLimits {
 	u64? max_total_opening_fee_msat;
 	u64? max_proportional_opening_fee_ppm_msat;
@@ -507,6 +520,55 @@ dictionary PaymentDetails {
 	u64 latest_update_timestamp;
 };
 
+dictionary ForwardedPaymentDetails {
+	ForwardedPaymentId id;
+	ChannelId prev_channel_id;
+	ChannelId next_channel_id;
+	UserChannelId? prev_user_channel_id;
+	UserChannelId? next_user_channel_id;
+	PublicKey? prev_node_id;
+	PublicKey? next_node_id;
+	u64? total_fee_earned_msat;
+	u64? skimmed_fee_msat;
+	boolean claim_from_onchain_tx;
+	u64? outbound_amount_forwarded_msat;
+	u64 forwarded_at_timestamp;
+};
+
+dictionary ChannelForwardingStats {
+	ChannelId channel_id;
+	PublicKey? counterparty_node_id;
+	u64 inbound_payments_forwarded;
+	u64 outbound_payments_forwarded;
+	u64 total_inbound_amount_msat;
+	u64 total_outbound_amount_msat;
+	u64 total_fee_earned_msat;
+	u64 total_skimmed_fee_msat;
+	u64 onchain_claims_count;
+	u64 first_forwarded_at_timestamp;
+	u64 last_forwarded_at_timestamp;
+};
+
+dictionary ChannelPairForwardingStats {
+	ChannelPairStatsId id;
+	ChannelId prev_channel_id;
+	ChannelId next_channel_id;
+	u64 bucket_start_timestamp;
+	PublicKey? prev_node_id;
+	PublicKey? next_node_id;
+	u64 payment_count;
+	u64 total_inbound_amount_msat;
+	u64 total_outbound_amount_msat;
+	u64 total_fee_earned_msat;
+	u64 total_skimmed_fee_msat;
+	u64 onchain_claims_count;
+	u64 avg_fee_msat;
+	u64 avg_inbound_amount_msat;
+	u64 first_forwarded_at_timestamp;
+	u64 last_forwarded_at_timestamp;
+	u64 aggregated_at_timestamp;
+};
+
 dictionary RouteParametersConfig {
 	u64? max_total_routing_fee_msat;
 	u32 max_total_cltv_expiry_delta;
@@ -894,6 +956,9 @@ typedef string OfferId;
 [Custom]
 typedef string PaymentId;
 
+[Custom]
+typedef string ForwardedPaymentId;
+
 [Custom]
 typedef string PaymentHash;
 
@@ -906,6 +971,9 @@ typedef string PaymentSecret;
 [Custom]
 typedef string ChannelId;
 
+[Custom]
+typedef string ChannelPairStatsId;
+
 [Custom]
 typedef string UserChannelId;
 

src/builder.rs

@@ -55,13 +55,20 @@ use crate::fee_estimator::OnchainFeeEstimator;
 use crate::gossip::GossipSource;
 use crate::io::sqlite_store::SqliteStore;
 use crate::io::utils::{
-	read_event_queue, read_external_pathfinding_scores_from_cache, read_network_graph,
+	read_channel_forwarding_stats, read_channel_pair_forwarding_stats, read_event_queue,
+	read_external_pathfinding_scores_from_cache, read_forwarded_payments, read_network_graph,
 	read_node_metrics, read_output_sweeper, read_payments, read_peer_info, read_pending_payments,
 	read_scorer, write_node_metrics,
 };
 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,
 };
@@ -75,9 +82,10 @@ use crate::peer_store::PeerStore;
 use crate::runtime::{Runtime, RuntimeSpawner};
 use crate::tx_broadcaster::TransactionBroadcaster;
 use crate::types::{
-	AsyncPersister, ChainMonitor, ChannelManager, DynStore, DynStoreWrapper, GossipSync, Graph,
-	KeysManager, MessageRouter, OnionMessenger, PaymentStore, PeerManager, PendingPaymentStore,
-	Persister, SyncAndAsyncKVStore,
+	AsyncPersister, ChainMonitor, ChannelForwardingStatsStore, ChannelManager,
+	ChannelPairForwardingStatsStore, DynStore, DynStoreWrapper, ForwardedPaymentStore, GossipSync,
+	Graph, KeysManager, MessageRouter, OnionMessenger, PaymentStore, PeerManager,
+	PendingPaymentStore, Persister, SyncAndAsyncKVStore,
 };
 use crate::wallet::persist::KVStoreWalletPersister;
 use crate::wallet::Wallet;
@@ -1060,14 +1068,23 @@ fn build_with_store_internal(
 
 	let kv_store_ref = Arc::clone(&kv_store);
 	let logger_ref = Arc::clone(&logger);
-	let (payment_store_res, node_metris_res, pending_payment_store_res) =
-		runtime.block_on(async move {
-			tokio::join!(
-				read_payments(&*kv_store_ref, Arc::clone(&logger_ref)),
-				read_node_metrics(&*kv_store_ref, Arc::clone(&logger_ref)),
-				read_pending_payments(&*kv_store_ref, Arc::clone(&logger_ref))
-			)
-		});
+	let (
+		payment_store_res,
+		forwarded_payment_store_res,
+		channel_forwarding_stats_res,
+		channel_pair_forwarding_stats_res,
+		node_metris_res,
+		pending_payment_store_res,
+	) = runtime.block_on(async move {
+		tokio::join!(
+			read_payments(&*kv_store_ref, Arc::clone(&logger_ref)),
+			read_forwarded_payments(&*kv_store_ref, Arc::clone(&logger_ref)),
+			read_channel_forwarding_stats(&*kv_store_ref, Arc::clone(&logger_ref)),
+			read_channel_pair_forwarding_stats(&*kv_store_ref, Arc::clone(&logger_ref)),
+			read_node_metrics(&*kv_store_ref, Arc::clone(&logger_ref)),
+			read_pending_payments(&*kv_store_ref, Arc::clone(&logger_ref))
+		)
+	});
 
 	// Initialize the status fields.
 	let node_metrics = match node_metris_res {
@@ -1096,6 +1113,48 @@ fn build_with_store_internal(
 		},
 	};
 
+	let forwarded_payment_store = match forwarded_payment_store_res {
+		Ok(forwarded_payments) => Arc::new(ForwardedPaymentStore::new(
+			forwarded_payments,
+			FORWARDED_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE.to_string(),
+			FORWARDED_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE.to_string(),
+			Arc::clone(&kv_store),
+			Arc::clone(&logger),
+		)),
+		Err(e) => {
+			log_error!(logger, "Failed to read forwarded payment data from store: {}", e);
+			return Err(BuildError::ReadFailed);
+		},
+	};
+
+	let channel_forwarding_stats_store = match channel_forwarding_stats_res {
+		Ok(stats) => Arc::new(ChannelForwardingStatsStore::new(
+			stats,
+			CHANNEL_FORWARDING_STATS_PERSISTENCE_PRIMARY_NAMESPACE.to_string(),
+			CHANNEL_FORWARDING_STATS_PERSISTENCE_SECONDARY_NAMESPACE.to_string(),
+			Arc::clone(&kv_store),
+			Arc::clone(&logger),
+		)),
+		Err(e) => {
+			log_error!(logger, "Failed to read channel forwarding stats from store: {}", e);
+			return Err(BuildError::ReadFailed);
+		},
+	};
+
+	let channel_pair_forwarding_stats_store = match channel_pair_forwarding_stats_res {
+		Ok(stats) => Arc::new(ChannelPairForwardingStatsStore::new(
+			stats,
+			CHANNEL_PAIR_FORWARDING_STATS_PERSISTENCE_PRIMARY_NAMESPACE.to_string(),
+			CHANNEL_PAIR_FORWARDING_STATS_PERSISTENCE_SECONDARY_NAMESPACE.to_string(),
+			Arc::clone(&kv_store),
+			Arc::clone(&logger),
+		)),
+		Err(e) => {
+			log_error!(logger, "Failed to read channel pair forwarding stats from store: {}", e);
+			return Err(BuildError::ReadFailed);
+		},
+	};
+
 	let (chain_source, chain_tip_opt) = match chain_data_source_config {
 		Some(ChainDataSourceConfig::Esplora { server_url, headers, sync_config }) => {
 			let sync_config = sync_config.unwrap_or(EsploraSyncConfig::default());
@@ -1782,6 +1841,9 @@ fn build_with_store_internal(
 		scorer,
 		peer_store,
 		payment_store,
+		forwarded_payment_store,
+		channel_forwarding_stats_store,
+		channel_pair_forwarding_stats_store,
 		is_running,
 		node_metrics,
 		om_mailbox,

src/config.rs

@@ -21,6 +21,30 @@ use lightning::util::config::{
 
 use crate::logger::LogLevel;
 
+/// The mode used for tracking forwarded payments.
+///
+/// This determines how much detail is stored about payment forwarding activity.
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+pub enum ForwardedPaymentTrackingMode {
+	/// Store individual forwarded payments for the specified retention period (in minutes),
+	/// then aggregate into channel-pair statistics.
+	///
+	/// Individual payments older than `retention_minutes` are aggregated by channel pair
+	/// and removed. Set to 0 for unlimited retention.
+	Detailed {
+		/// The retention period for individual forwarded payment records, in minutes.
+		/// Individual payments older than this period are aggregated into channel-pair statistics and removed.
+		/// Set to 0 for unlimited retention.
+		retention_minutes: u64,
+	},
+	/// Track only per-channel aggregate statistics without storing individual payment records.
+	///
+	/// This is the default mode. Use this to reduce storage requirements when you only need
+	/// aggregate metrics like total fees earned per channel.
+	#[default]
+	Stats,
+}
+
 // Config defaults
 const DEFAULT_NETWORK: Network = Network::Bitcoin;
 const DEFAULT_BDK_WALLET_SYNC_INTERVAL_SECS: u64 = 80;
@@ -127,9 +151,10 @@ pub(crate) const HRN_RESOLUTION_TIMEOUT_SECS: u64 = 5;
 /// | `probing_liquidity_limit_multiplier`   | 3                  |
 /// | `log_level`                            | Debug              |
 /// | `anchor_channels_config`               | Some(..)           |
-/// | `route_parameters`                   | None               |
+/// | `route_parameters`                     | None               |
+/// | `forwarded_payment_tracking_mode`      | Detailed           |
 ///
-/// See [`AnchorChannelsConfig`] and [`RouteParametersConfig`] for more information regarding their
+/// See [`AnchorChannelsConfig`], [`RouteParametersConfig`], and [`ForwardedPaymentTrackingMode`] for more information regarding their
 /// respective default values.
 ///
 /// [`Node`]: crate::Node
@@ -192,6 +217,10 @@ pub struct Config {
 	/// **Note:** If unset, default parameters will be used, and you will be able to override the
 	/// parameters on a per-payment basis in the corresponding method calls.
 	pub route_parameters: Option<RouteParametersConfig>,
+	/// The mode used for tracking forwarded payments.
+	///
+	/// See [`ForwardedPaymentTrackingMode`] for more information on the available modes.
+	pub forwarded_payment_tracking_mode: ForwardedPaymentTrackingMode,
 }
 
 impl Default for Config {
@@ -206,6 +235,7 @@ impl Default for Config {
 			anchor_channels_config: Some(AnchorChannelsConfig::default()),
 			route_parameters: None,
 			node_alias: None,
+			forwarded_payment_tracking_mode: ForwardedPaymentTrackingMode::default(),
 		}
 	}
 }