Add per-channel forwarding statistics

Track aggregate stats (fees earned, payment counts, amounts) per channel.
Add ForwardedPaymentTrackingMode config: Stats (default) for lightweight
metrics only, or Detailed to also store individual payment records.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: benthecarman

src/builder.rs

@@ -55,13 +55,15 @@ 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_forwarded_payments,
-	read_network_graph, read_node_metrics, read_output_sweeper, read_payments, read_peer_info,
-	read_pending_payments, read_scorer, write_node_metrics,
+	read_channel_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, FORWARDED_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
+	self, CHANNEL_FORWARDING_STATS_PERSISTENCE_PRIMARY_NAMESPACE,
+	CHANNEL_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,
@@ -77,9 +79,10 @@ use crate::peer_store::PeerStore;
 use crate::runtime::{Runtime, RuntimeSpawner};
 use crate::tx_broadcaster::TransactionBroadcaster;
 use crate::types::{
-	AsyncPersister, ChainMonitor, ChannelManager, DynStore, DynStoreWrapper, ForwardedPaymentStore,
-	GossipSync, Graph, KeysManager, MessageRouter, OnionMessenger, PaymentStore, PeerManager,
-	PendingPaymentStore, Persister, SyncAndAsyncKVStore,
+	AsyncPersister, ChainMonitor, ChannelForwardingStatsStore, ChannelManager, DynStore,
+	DynStoreWrapper, ForwardedPaymentStore, GossipSync, Graph, KeysManager, MessageRouter,
+	OnionMessenger, PaymentStore, PeerManager, PendingPaymentStore, Persister,
+	SyncAndAsyncKVStore,
 };
 use crate::wallet::persist::KVStoreWalletPersister;
 use crate::wallet::Wallet;
@@ -1066,12 +1069,14 @@ fn build_with_store_internal(
 	let (
 		payment_store_res,
 		forwarded_payment_store_res,
+		channel_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_node_metrics(&*kv_store_ref, Arc::clone(&logger_ref)),
 			read_pending_payments(&*kv_store_ref, Arc::clone(&logger_ref))
 		)
@@ -1118,6 +1123,20 @@ fn build_with_store_internal(
 		},
 	};
 
+	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 (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());
@@ -1805,6 +1824,7 @@ fn build_with_store_internal(
 		peer_store,
 		payment_store,
 		forwarded_payment_store,
+		channel_forwarding_stats_store,
 		is_running,
 		node_metrics,
 		om_mailbox,

src/config.rs

@@ -21,6 +21,24 @@ 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 every individual forwarded payment AND track per-channel aggregate statistics.
+	///
+	/// Use this when you need full history of forwarded payments for accounting, debugging,
+	/// or detailed analytics.
+	Detailed,
+	/// 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 +145,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 +211,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 +229,7 @@ impl Default for Config {
 			anchor_channels_config: Some(AnchorChannelsConfig::default()),
 			route_parameters: None,
 			node_alias: None,
+			forwarded_payment_tracking_mode: ForwardedPaymentTrackingMode::default(),
 		}
 	}
 }

src/event.rs

@@ -33,7 +33,7 @@ use lightning_liquidity::lsps2::utils::compute_opening_fee;
 use lightning_types::payment::{PaymentHash, PaymentPreimage};
 use rand::{rng, Rng};
 
-use crate::config::{may_announce_channel, Config};
+use crate::config::{may_announce_channel, Config, ForwardedPaymentTrackingMode};
 use crate::connection::ConnectionManager;
 use crate::data_store::DataStoreUpdateResult;
 use crate::fee_estimator::ConfirmationTarget;
@@ -46,12 +46,13 @@ use crate::logger::{log_debug, log_error, log_info, log_trace, LdkLogger, Logger
 use crate::payment::asynchronous::om_mailbox::OnionMessageMailbox;
 use crate::payment::asynchronous::static_invoice_store::StaticInvoiceStore;
 use crate::payment::store::{
-	ForwardedPaymentDetails, ForwardedPaymentId, PaymentDetails, PaymentDetailsUpdate,
-	PaymentDirection, PaymentKind, PaymentStatus,
+	ChannelForwardingStats, ForwardedPaymentDetails, ForwardedPaymentId, PaymentDetails,
+	PaymentDetailsUpdate, PaymentDirection, PaymentKind, PaymentStatus,
 };
 use crate::runtime::Runtime;
 use crate::types::{
-	CustomTlvRecord, DynStore, ForwardedPaymentStore, OnionMessenger, PaymentStore, Sweeper, Wallet,
+	ChannelForwardingStatsStore, CustomTlvRecord, DynStore, ForwardedPaymentStore, OnionMessenger,
+	PaymentStore, Sweeper, Wallet,
 };
 use crate::{
 	hex_utils, BumpTransactionEventHandler, ChannelManager, Error, Graph, PeerInfo, PeerStore,
@@ -492,6 +493,7 @@ where
 	liquidity_source: Option<Arc<LiquiditySource<Arc<Logger>>>>,
 	payment_store: Arc<PaymentStore>,
 	forwarded_payment_store: Arc<ForwardedPaymentStore>,
+	channel_forwarding_stats_store: Arc<ChannelForwardingStatsStore>,
 	peer_store: Arc<PeerStore<L>>,
 	runtime: Arc<Runtime>,
 	logger: L,
@@ -512,6 +514,7 @@ where
 		output_sweeper: Arc<Sweeper>, network_graph: Arc<Graph>,
 		liquidity_source: Option<Arc<LiquiditySource<Arc<Logger>>>>,
 		payment_store: Arc<PaymentStore>, forwarded_payment_store: Arc<ForwardedPaymentStore>,
+		channel_forwarding_stats_store: Arc<ChannelForwardingStatsStore>,
 		peer_store: Arc<PeerStore<L>>, static_invoice_store: Option<StaticInvoiceStore>,
 		onion_messenger: Arc<OnionMessenger>, om_mailbox: Option<Arc<OnionMessageMailbox>>,
 		runtime: Arc<Runtime>, logger: L, config: Arc<Config>,
@@ -527,6 +530,7 @@ where
 			liquidity_source,
 			payment_store,
 			forwarded_payment_store,
+			channel_forwarding_stats_store,
 			peer_store,
 			logger,
 			runtime,
@@ -1370,40 +1374,99 @@ where
 						.await;
 				}
 
-				// Store the forwarded payment details
 				let prev_channel_id_value = prev_channel_id
 					.expect("prev_channel_id expected for events generated by LDK versions greater than 0.0.107.");
 				let next_channel_id_value = next_channel_id
 					.expect("next_channel_id expected for events generated by LDK versions greater than 0.0.107.");
 
-				// PaymentForwarded does not have a unique id, so we generate a random one here.
-				let mut id_bytes = [0u8; 32];
-				rng().fill(&mut id_bytes);
-
 				let forwarded_at_timestamp = SystemTime::now()
 					.duration_since(UNIX_EPOCH)
 					.expect("SystemTime::now() should come after SystemTime::UNIX_EPOCH")
 					.as_secs();
 
-				let forwarded_payment = ForwardedPaymentDetails {
-					id: ForwardedPaymentId(id_bytes),
-					prev_channel_id: prev_channel_id_value,
-					next_channel_id: next_channel_id_value,
-					prev_user_channel_id: prev_user_channel_id.map(UserChannelId),
-					next_user_channel_id: next_user_channel_id.map(UserChannelId),
-					prev_node_id,
-					next_node_id,
-					total_fee_earned_msat,
-					skimmed_fee_msat,
-					claim_from_onchain_tx,
-					outbound_amount_forwarded_msat,
-					forwarded_at_timestamp,
+				// Calculate inbound amount (outbound + fee)
+				let inbound_amount_msat = outbound_amount_forwarded_msat
+					.unwrap_or(0)
+					.saturating_add(total_fee_earned_msat.unwrap_or(0));
+
+				// Update per-channel forwarding stats for the inbound channel (prev_channel)
+				// For new entries, this becomes the initial value; for existing entries,
+				// these values are used as increments via the to_update() -> update() pattern.
+				let inbound_stats = ChannelForwardingStats {
+					channel_id: prev_channel_id_value,
+					counterparty_node_id: prev_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: total_fee_earned_msat.unwrap_or(0),
+					total_skimmed_fee_msat: skimmed_fee_msat.unwrap_or(0),
+					onchain_claims_count: if claim_from_onchain_tx { 1 } else { 0 },
+					first_forwarded_at_timestamp: forwarded_at_timestamp,
+					last_forwarded_at_timestamp: forwarded_at_timestamp,
+				};
+				self.channel_forwarding_stats_store
+					.insert_or_update(inbound_stats)
+					.map_err(|e| {
+						log_error!(
+							self.logger,
+							"Failed to update inbound channel forwarding stats: {e}"
+						);
+						ReplayEvent()
+					})?;
+
+				// Update per-channel forwarding stats for the outbound channel (next_channel)
+				let outbound_stats = ChannelForwardingStats {
+					channel_id: next_channel_id_value,
+					counterparty_node_id: next_node_id,
+					inbound_payments_forwarded: 0,
+					outbound_payments_forwarded: 1,
+					total_inbound_amount_msat: 0,
+					total_outbound_amount_msat: outbound_amount_forwarded_msat.unwrap_or(0),
+					total_fee_earned_msat: 0,
+					total_skimmed_fee_msat: 0,
+					onchain_claims_count: 0,
+					first_forwarded_at_timestamp: forwarded_at_timestamp,
+					last_forwarded_at_timestamp: forwarded_at_timestamp,
 				};
+				self.channel_forwarding_stats_store
+					.insert_or_update(outbound_stats)
+					.map_err(|e| {
+						log_error!(
+							self.logger,
+							"Failed to update outbound channel forwarding stats: {e}"
+						);
+						ReplayEvent()
+					})?;
 
-				self.forwarded_payment_store.insert(forwarded_payment).map_err(|e| {
-					log_error!(self.logger, "Failed to store forwarded payment: {e}");
-					ReplayEvent()
-				})?;
+				// Only store individual forwarded payment details in Detailed mode
+				if self.config.forwarded_payment_tracking_mode
+					== ForwardedPaymentTrackingMode::Detailed
+				{
+					// PaymentForwarded does not have a unique id, so we generate a random one here.
+					let mut id_bytes = [0u8; 32];
+					rng().fill(&mut id_bytes);
+
+					let forwarded_payment = ForwardedPaymentDetails {
+						id: ForwardedPaymentId(id_bytes),
+						prev_channel_id: prev_channel_id_value,
+						next_channel_id: next_channel_id_value,
+						prev_user_channel_id: prev_user_channel_id.map(UserChannelId),
+						next_user_channel_id: next_user_channel_id.map(UserChannelId),
+						prev_node_id,
+						next_node_id,
+						total_fee_earned_msat,
+						skimmed_fee_msat,
+						claim_from_onchain_tx,
+						outbound_amount_forwarded_msat,
+						forwarded_at_timestamp,
+					};
+
+					self.forwarded_payment_store.insert(forwarded_payment).map_err(|e| {
+						log_error!(self.logger, "Failed to store forwarded payment: {e}");
+						ReplayEvent()
+					})?;
+				}
 
 				let event = Event::PaymentForwarded {
 					prev_channel_id: prev_channel_id_value,

src/io/mod.rs

@@ -31,6 +31,11 @@ pub(crate) const PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE: &str = "";
 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 node metrics will be persisted under this key.
 pub(crate) const NODE_METRICS_PRIMARY_NAMESPACE: &str = "";
 pub(crate) const NODE_METRICS_SECONDARY_NAMESPACE: &str = "";

src/io/utils.rs

@@ -39,14 +39,18 @@ use rand::rngs::OsRng;
 use rand::TryRngCore;
 
 use super::*;
+use crate::io::{
+	CHANNEL_FORWARDING_STATS_PERSISTENCE_PRIMARY_NAMESPACE,
+	CHANNEL_FORWARDING_STATS_PERSISTENCE_SECONDARY_NAMESPACE,
+};
 use crate::chain::ChainSource;
 use crate::config::WALLET_KEYS_SEED_LEN;
 use crate::fee_estimator::OnchainFeeEstimator;
 use crate::io::{
 	NODE_METRICS_KEY, NODE_METRICS_PRIMARY_NAMESPACE, NODE_METRICS_SECONDARY_NAMESPACE,
 };
 use crate::logger::{log_error, LdkLogger, Logger};
-use crate::payment::{ForwardedPaymentDetails, PendingPaymentDetails};
+use crate::payment::{ChannelForwardingStats, ForwardedPaymentDetails, PendingPaymentDetails};
 use crate::peer_store::PeerStore;
 use crate::types::{Broadcaster, DynStore, KeysManager, Sweeper};
 use crate::wallet::ser::{ChangeSetDeserWrapper, ChangeSetSerWrapper};
@@ -320,6 +324,22 @@ where
 	.await
 }
 
+/// Read previously persisted channel forwarding stats from the store.
+pub(crate) async fn read_channel_forwarding_stats<L: Deref>(
+	kv_store: &DynStore, logger: L,
+) -> Result<Vec<ChannelForwardingStats>, std::io::Error>
+where
+	L::Target: LdkLogger,
+{
+	read_objects_from_store(
+		kv_store,
+		logger,
+		CHANNEL_FORWARDING_STATS_PERSISTENCE_PRIMARY_NAMESPACE,
+		CHANNEL_FORWARDING_STATS_PERSISTENCE_SECONDARY_NAMESPACE,
+	)
+	.await
+}
+
 /// Read `OutputSweeper` state from the store.
 pub(crate) async fn read_output_sweeper(
 	broadcaster: Arc<Broadcaster>, fee_estimator: Arc<OnchainFeeEstimator>,