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. Keep forwarding persistence
and event-recording logic behind one internal store.

AI-assisted-by: OpenAI Codex and Anthropic Fable

Author: benthecarman

bindings/ldk_node.udl

@@ -10,6 +10,8 @@ typedef dictionary ElectrumSyncConfig;
 
 typedef dictionary TorConfig;
 
+typedef enum ForwardedPaymentTrackingMode;
+
 typedef interface NodeEntropy;
 
 typedef interface ProbingConfig;
@@ -68,6 +70,7 @@ interface Node {
 	SpontaneousPayment spontaneous_payment();
 	OnchainPayment onchain_payment();
 	Liquidity liquidity();
+	ForwardingAnalytics forwarding_analytics();
 	[Throws=NodeError]
 	void lnurl_auth(string lnurl);
 	[Throws=NodeError]
@@ -139,6 +142,8 @@ interface FeeRate {
 
 typedef interface Liquidity;
 
+typedef interface ForwardingAnalytics;
+
 [Error]
 enum NodeError {
 	"AlreadyRunning",
@@ -177,6 +182,8 @@ enum NodeError {
 	"InvalidOfferId",
 	"InvalidNodeId",
 	"InvalidPaymentId",
+	"InvalidForwardedPaymentId",
+	"InvalidChannelPairForwardingStatsId",
 	"InvalidPaymentHash",
 	"InvalidPaymentPreimage",
 	"InvalidPaymentSecret",
@@ -358,6 +365,12 @@ typedef string OfferId;
 [Custom]
 typedef string PaymentId;
 
+[Custom]
+typedef string ForwardedPaymentId;
+
+[Custom]
+typedef string ChannelPairForwardingStatsId;
+
 [Custom]
 typedef string PaymentHash;
 
@@ -395,3 +408,15 @@ typedef enum Event;
 typedef interface HRNResolverConfig;
 
 typedef dictionary HumanReadableNamesConfig;
+
+typedef dictionary ForwardedPaymentDetails;
+
+typedef dictionary ChannelForwardingStats;
+
+typedef dictionary ChannelPairForwardingStats;
+
+typedef dictionary ForwardedPaymentDetailsPage;
+
+typedef dictionary ChannelForwardingStatsPage;
+
+typedef dictionary ChannelPairForwardingStatsPage;

src/builder.rs

@@ -80,7 +80,9 @@ use crate::io::utils::{
 #[cfg(feature = "storage-vss")]
 use crate::io::vss_store::VssStoreBuilder;
 use crate::io::{
-	self, PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
+	self, CHANNEL_FORWARDING_STATS_PERSISTENCE_SECONDARY_NAMESPACE,
+	FORWARDED_PAYMENT_PERSISTENCE_PRIMARY_NAMESPACE, PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
+	PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
 	PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
 	PENDING_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
 };
@@ -89,6 +91,7 @@ use crate::lnurl_auth::LnurlAuth;
 use crate::logger::{log_error, LdkLogger, LogLevel, LogWriter, Logger};
 use crate::message_handler::NodeCustomMessageHandler;
 use crate::payment::asynchronous::om_mailbox::OnionMessageMailbox;
+use crate::payment::forwarding_store::ForwardingStore;
 #[cfg(feature = "unified-payments")]
 use crate::payment::HRNResolver;
 use crate::peer_store::PeerStore;
@@ -1524,26 +1527,37 @@ fn build_with_store_internal(
 
 	let kv_store_ref = Arc::clone(&kv_store);
 	let logger_ref = Arc::clone(&logger);
-	let (payment_store_res, node_metris_res, pending_payment_store_res, address_pool_res) = runtime
-		.block_on(async move {
-			tokio::join!(
-				read_n_objects(
-					&*kv_store_ref,
-					PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
-					PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
-					PAYMENT_CACHE_WARMUP_COUNT,
-					Arc::clone(&logger_ref),
-				),
-				read_node_metrics(&*kv_store_ref, Arc::clone(&logger_ref)),
-				read_all_objects(
-					&*kv_store_ref,
-					PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
-					PENDING_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
-					Arc::clone(&logger_ref),
-				),
-				read_address_pool(&*kv_store_ref, &*logger_ref)
-			)
-		});
+	let (
+		payment_store_res,
+		channel_forwarding_stats_res,
+		node_metris_res,
+		pending_payment_store_res,
+		address_pool_res,
+	) = runtime.block_on(async move {
+		tokio::join!(
+			read_n_objects(
+				&*kv_store_ref,
+				PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
+				PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
+				PAYMENT_CACHE_WARMUP_COUNT,
+				Arc::clone(&logger_ref),
+			),
+			read_all_objects(
+				&*kv_store_ref,
+				FORWARDED_PAYMENT_PERSISTENCE_PRIMARY_NAMESPACE,
+				CHANNEL_FORWARDING_STATS_PERSISTENCE_SECONDARY_NAMESPACE,
+				Arc::clone(&logger_ref),
+			),
+			read_node_metrics(&*kv_store_ref, Arc::clone(&logger_ref)),
+			read_all_objects(
+				&*kv_store_ref,
+				PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
+				PENDING_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
+				Arc::clone(&logger_ref),
+			),
+			read_address_pool(&*kv_store_ref, &*logger_ref),
+		)
+	});
 
 	// Initialize the status fields.
 	let node_metrics = match node_metris_res {
@@ -1576,6 +1590,14 @@ fn build_with_store_internal(
 		},
 	};
 
+	let channel_forwarding_stats = match channel_forwarding_stats_res {
+		Ok(stats) => stats,
+		Err(e) => {
+			log_error!(logger, "Failed to read channel forwarding stats from store: {}", e);
+			return Err(BuildError::ReadFailed);
+		},
+	};
+
 	let (chain_source, chain_tip_opt) = match chain_data_source_config {
 		#[cfg(feature = "chain-esplora")]
 		Some(ChainDataSourceConfig::Esplora { server_url, headers, sync_config }) => {
@@ -1902,6 +1924,12 @@ fn build_with_store_internal(
 		Arc::clone(&wallet),
 		Arc::clone(&logger),
 	));
+	let forwarding_store = Arc::new(ForwardingStore::new(
+		channel_forwarding_stats,
+		config.forwarded_payment_tracking_mode,
+		Arc::clone(&kv_store),
+		Arc::clone(&logger),
+	));
 
 	let peer_storage_key = keys_manager.get_peer_storage_key();
 	let monitor_reader = Arc::new(AsyncPersister::new(
@@ -2457,6 +2485,16 @@ fn build_with_store_internal(
 		_leak_checker.0.push(Arc::downgrade(&wallet) as Weak<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,
@@ -2484,6 +2522,8 @@ fn build_with_store_internal(
 		scorer,
 		peer_store,
 		payment_store,
+		forwarding_store,
+		forwarded_payment_aggregation_retention_secs,
 		lnurl_auth,
 		is_running,
 		node_metrics,

src/config.rs

@@ -169,6 +169,30 @@ pub(crate) const LIQUIDITY_DISCOVERY_RETRY_INITIAL_DELAY: Duration = Duration::f
 // thereafter until every configured LSP has been discovered.
 pub(crate) const LIQUIDITY_DISCOVERY_RETRY_MAX_DELAY: Duration = Duration::from_secs(60 * 60);
 
+/// The mode used for tracking forwarded payments.
+///
+/// In either mode, a forward is tracked only when it has exactly one incoming HTLC and one outgoing
+/// HTLC, and LDK reports both the outbound amount and total fee.
+#[derive(Copy, Clone, Debug, PartialEq, Eq)]
+#[cfg_attr(feature = "uniffi", derive(uniffi::Enum))]
+pub enum ForwardedPaymentTrackingMode {
+	/// Track eligible new forwarded payments only as per-channel aggregate statistics.
+	///
+	/// Any detailed records left by a previous configuration are aggregated and removed after their
+	/// current one-hour bucket closes.
+	Stats,
+	/// Store eligible individual forwarded payments for the current and previous one-hour buckets.
+	///
+	/// Payments from older buckets are aggregated into channel-pair statistics and removed.
+	Detailed,
+}
+
+impl Default for ForwardedPaymentTrackingMode {
+	fn default() -> Self {
+		Self::Stats
+	}
+}
+
 #[derive(Debug, Clone)]
 #[cfg_attr(feature = "uniffi", derive(uniffi::Record))]
 /// Represents the configuration of an [`Node`] instance.
@@ -192,9 +216,10 @@ pub(crate) const LIQUIDITY_DISCOVERY_RETRY_MAX_DELAY: Duration = Duration::from_
 	doc = "| `hrn_config`                           | HumanReadableNamesConfig::default()  |"
 )]
 /// | `manually_handle_unknown_bolt11_payments` | false                              |
+/// | `forwarded_payment_tracking_mode`      | Stats                               |
 ///
-/// See [`AnchorChannelsConfig`] and [`RouteParametersConfig`] for more information regarding their
-/// respective default values.
+/// See [`AnchorChannelsConfig`], [`RouteParametersConfig`], and
+/// [`ForwardedPaymentTrackingMode`] for more information regarding their respective default values.
 ///
 /// [`Node`]: crate::Node
 pub struct Config {
@@ -268,6 +293,8 @@ pub struct Config {
 	///
 	/// [`Event::PaymentClaimable`]: crate::Event::PaymentClaimable
 	pub manually_handle_unknown_bolt11_payments: bool,
+	/// The mode used for tracking forwarded payments.
+	pub forwarded_payment_tracking_mode: ForwardedPaymentTrackingMode,
 }
 
 impl Default for Config {
@@ -286,6 +313,7 @@ impl Default for Config {
 			#[cfg(feature = "unified-payments")]
 			hrn_config: HumanReadableNamesConfig::default(),
 			manually_handle_unknown_bolt11_payments: false,
+			forwarded_payment_tracking_mode: ForwardedPaymentTrackingMode::default(),
 		}
 	}
 }

src/error.rs

@@ -89,6 +89,10 @@ pub enum Error {
 	InvalidNodeId,
 	/// The given payment id is invalid.
 	InvalidPaymentId,
+	/// The given forwarded payment id is invalid.
+	InvalidForwardedPaymentId,
+	/// The given channel-pair forwarding statistics id is invalid.
+	InvalidChannelPairForwardingStatsId,
 	/// The given payment hash is invalid.
 	InvalidPaymentHash,
 	/// The given payment pre-image is invalid.
@@ -194,6 +198,12 @@ impl fmt::Display for Error {
 			Self::InvalidOfferId => write!(f, "The given offer id is invalid."),
 			Self::InvalidNodeId => write!(f, "The given node id is invalid."),
 			Self::InvalidPaymentId => write!(f, "The given payment id is invalid."),
+			Self::InvalidForwardedPaymentId => {
+				write!(f, "The given forwarded payment id is invalid.")
+			},
+			Self::InvalidChannelPairForwardingStatsId => {
+				write!(f, "The given channel-pair forwarding statistics id is invalid.")
+			},
 			Self::InvalidPaymentHash => write!(f, "The given payment hash is invalid."),
 			Self::InvalidPaymentPreimage => write!(f, "The given payment preimage is invalid."),
 			Self::InvalidPaymentSecret => write!(f, "The given payment secret is invalid."),