Add forwarded payment tracking

Store unambiguous single-HTLC forwarding events. Aggregate them into
per-channel and channel-pair statistics.

Use fixed one-hour buckets for detailed records. Keep details out of the
payment LRU cache. Use one persistence namespace for forwarding data.

Expose analytics through Rust and UniFFI. Preserve raw details if an
aggregate marker cannot be decoded.

AI-assisted-by: OpenAI Codex and Anthropic Fable

Author: benthecarman

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();
+	ForwardingAnalytics forwarding_analytics();
 	[Throws=NodeError]
 	void lnurl_auth(string lnurl);
 	[Throws=NodeError]
@@ -186,6 +189,8 @@ typedef interface UnifiedPayment;
 
 typedef interface Liquidity;
 
+typedef interface ForwardingAnalytics;
+
 [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;

src/builder.rs

@@ -54,7 +54,7 @@ use crate::config::{
 	PAYMENT_CACHE_WARMUP_COUNT,
 };
 use crate::connection::ConnectionManager;
-use crate::data_store::{KeepAllEntries, KeepLeastRecentlyUsed};
+use crate::data_store::{KeepAllEntries, KeepLeastRecentlyUsed, KeepNoEntries};
 use crate::entropy::NodeEntropy;
 use crate::event::EventQueue;
 use crate::fee_estimator::OnchainFeeEstimator;
@@ -67,7 +67,11 @@ 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_SECONDARY_NAMESPACE,
+	CHANNEL_PAIR_FORWARDING_STATS_PERSISTENCE_SECONDARY_NAMESPACE,
+	FORWARDED_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
+	FORWARDED_PAYMENT_PERSISTENCE_PRIMARY_NAMESPACE, PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
+	PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
 	PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
 	PENDING_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
 };
@@ -84,7 +88,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 +1462,37 @@ fn build_with_store_internal(
 
 	let kv_store_ref = Arc::clone(&kv_store);
 	let logger_ref = Arc::clone(&logger);
-	let (payment_store_res, node_metris_res, pending_payment_store_res, address_pool_res) = runtime
-		.block_on(async move {
-			tokio::join!(
-				read_n_objects(
-					&*kv_store_ref,
-					PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
-					PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
-					PAYMENT_CACHE_WARMUP_COUNT,
-					Arc::clone(&logger_ref),
-				),
-				read_node_metrics(&*kv_store_ref, Arc::clone(&logger_ref)),
-				read_all_objects(
-					&*kv_store_ref,
-					PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
-					PENDING_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
-					Arc::clone(&logger_ref),
-				),
-				read_address_pool(&*kv_store_ref, &*logger_ref)
-			)
-		});
+	let (
+		payment_store_res,
+		channel_forwarding_stats_res,
+		node_metris_res,
+		pending_payment_store_res,
+		address_pool_res,
+	) = runtime.block_on(async move {
+		tokio::join!(
+			read_n_objects(
+				&*kv_store_ref,
+				PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
+				PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
+				PAYMENT_CACHE_WARMUP_COUNT,
+				Arc::clone(&logger_ref),
+			),
+			read_all_objects(
+				&*kv_store_ref,
+				FORWARDED_PAYMENT_PERSISTENCE_PRIMARY_NAMESPACE,
+				CHANNEL_FORWARDING_STATS_PERSISTENCE_SECONDARY_NAMESPACE,
+				Arc::clone(&logger_ref),
+			),
+			read_node_metrics(&*kv_store_ref, Arc::clone(&logger_ref)),
+			read_all_objects(
+				&*kv_store_ref,
+				PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
+				PENDING_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
+				Arc::clone(&logger_ref),
+			),
+			read_address_pool(&*kv_store_ref, &*logger_ref),
+		)
+	});
 
 	// Initialize the status fields.
 	let node_metrics = match node_metris_res {
@@ -1509,6 +1525,39 @@ fn build_with_store_internal(
 		},
 	};
 
+	let forwarded_payment_store = Arc::new(ForwardedPaymentStore::new(
+		Vec::new(),
+		KeepNoEntries,
+		FORWARDED_PAYMENT_PERSISTENCE_PRIMARY_NAMESPACE.to_string(),
+		FORWARDED_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE.to_string(),
+		Arc::clone(&kv_store),
+		Arc::clone(&logger),
+	));
+
+	let channel_forwarding_stats_store = match channel_forwarding_stats_res {
+		Ok(stats) => Arc::new(ChannelForwardingStatsStore::new(
+			stats,
+			KeepAllEntries,
+			FORWARDED_PAYMENT_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(
+		Vec::new(),
+		KeepNoEntries,
+		FORWARDED_PAYMENT_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 +2408,16 @@ fn build_with_store_internal(
 		_leak_checker.0.push(Arc::downgrade(&wallet) as Weak<dyn Any + Send + Sync>);
 	}
 
+	// 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 +2445,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,

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(),
 		}
 	}
 }