Add forwarded payment tracking and statistics

Routing nodes and LSPs need to track forwarded payments so they can
account for earned fees and profitability over time. Store forwarded
payment data and aggregate it into channel and channel-pair statistics.

Detailed tracking retains individual forwarding events for the current
and previous configured buckets, then aggregates older events into
channel-pair statistics. Persist the first detailed bucket length and
reject later mismatches so bucket geometry stays consistent across
restarts and mode changes. Stats mode records new forwards only in
per-channel totals and uses the persisted length to drain older details.

Persist channel-pair bucket keys in reversible URL-safe Base64. This
keeps them under LDK's KVStore limit without losing direct lookup or
namespace scanning. Treat persisted buckets as commit markers so retries
can finish detail cleanup without counting the same bucket twice.

Use per-HTLC locator amounts to attribute trampoline forwards across
channel pairs. Pair multi-HTLC forwards ordinally instead of inventing
a Cartesian product. When one side has extra HTLCs, reuse its final
entry and consume its amount across the remaining pairs in order. This
keeps totals exact without inventing proportional attribution.

Expose detected trampoline routing on forwarded payment events. Persist
the signal and derive it for older queued events from their stored HTLC
locator collections.

Preserve unknown fees in channel and channel-pair aggregates instead of
treating them as zero. Serialize detail insertion with bucket closure
and page reads with removals so concurrent retention work cannot drop
records or invalidate a page.

AI-assisted-by: OpenAI Codex

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;
@@ -152,6 +154,19 @@ interface Node {
 	void remove_payment([ByRef]PaymentId payment_id);
 	BalanceDetails list_balances();
 	sequence<PaymentDetails> list_payments();
+	ForwardedPaymentDetails? forwarded_payment([ByRef]ForwardedPaymentId forwarded_payment_id);
+	[Throws=NodeError]
+	ForwardedPaymentDetailsPage list_forwarded_payments(PageToken? page_token);
+	ForwardedPaymentTrackingMode forwarded_payment_tracking_mode();
+	ChannelForwardingStats? channel_forwarding_stats([ByRef]ChannelId channel_id);
+	[Throws=NodeError]
+	ChannelForwardingStatsPage list_channel_forwarding_stats(PageToken? page_token);
+	[Throws=NodeError]
+	ChannelPairForwardingStatsPage list_channel_pair_forwarding_stats(PageToken? page_token);
+	[Throws=NodeError]
+	ChannelPairForwardingStatsPage list_channel_pair_forwarding_stats_in_range(u64 start_timestamp, u64 end_timestamp, PageToken? page_token);
+	[Throws=NodeError]
+	ChannelPairForwardingStatsPage list_channel_pair_forwarding_stats_for_pair(ChannelId prev_channel_id, ChannelId next_channel_id, PageToken? page_token);
 	sequence<PeerDetails> list_peers();
 	sequence<ChannelDetails> list_channels();
 	NetworkGraph network_graph();
@@ -394,6 +409,9 @@ typedef string OfferId;
 [Custom]
 typedef string PaymentId;
 
+[Custom]
+typedef string ForwardedPaymentId;
+
 [Custom]
 typedef string PaymentHash;
 
@@ -406,6 +424,12 @@ typedef string PaymentSecret;
 [Custom]
 typedef string ChannelId;
 
+[Custom]
+typedef string ChannelPairStatsId;
+
+[Custom]
+typedef string PageToken;
+
 [Custom]
 typedef string UserChannelId;
 
@@ -432,3 +456,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

@@ -49,8 +49,9 @@ use crate::chain::ChainSource;
 use crate::config::{
 	default_user_config, may_announce_channel, AnnounceError, AsyncPaymentsRole,
 	BitcoindRestClientConfig, Config, ElectrumSyncConfig, EsploraSyncConfig, HRNResolverConfig,
-	TorConfig, DEFAULT_ESPLORA_SERVER_URL, DEFAULT_LOG_FILENAME, DEFAULT_LOG_LEVEL,
-	DEFAULT_MAX_PROBE_AMOUNT_MSAT, DEFAULT_MIN_PROBE_AMOUNT_MSAT,
+	ResolvedForwardedPaymentBucketConfig, TorConfig, DEFAULT_ESPLORA_SERVER_URL,
+	DEFAULT_LOG_FILENAME, DEFAULT_LOG_LEVEL, DEFAULT_MAX_PROBE_AMOUNT_MSAT,
+	DEFAULT_MIN_PROBE_AMOUNT_MSAT,
 };
 use crate::connection::ConnectionManager;
 use crate::entropy::NodeEntropy;
@@ -60,12 +61,19 @@ use crate::gossip::GossipSource;
 use crate::io::sqlite_store::SqliteStore;
 use crate::io::utils::{
 	open_or_migrate_fs_store, read_all_objects, read_event_queue,
-	read_external_pathfinding_scores_from_cache, read_network_graph, read_node_metrics,
-	read_output_sweeper, read_peer_info, read_scorer,
+	read_external_pathfinding_scores_from_cache, read_forwarded_payment_bucket_size,
+	read_network_graph, read_node_metrics, read_output_sweeper, read_peer_info, read_scorer,
+	write_forwarded_payment_bucket_size,
 };
 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,
 };
@@ -82,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,
 };
@@ -174,6 +183,10 @@ pub enum BuildError {
 	InvalidTorProxyAddress,
 	/// The provided alias is invalid.
 	InvalidNodeAlias,
+	/// The forwarded-payment bucket length is zero or too large to represent in seconds.
+	InvalidForwardedPaymentBucketLength,
+	/// The forwarded-payment bucket length differs from the value previously persisted.
+	ForwardedPaymentBucketLengthMismatch,
 	/// An attempt to setup a runtime has failed.
 	RuntimeSetupFailed,
 	/// We failed to read data from the [`KVStore`].
@@ -236,6 +249,12 @@ impl fmt::Display for BuildError {
 			Self::LoggerSetupFailed => write!(f, "Failed to setup the logger."),
 			Self::ChainSourceSetupFailed => write!(f, "Failed to setup the chain source."),
 			Self::InvalidNodeAlias => write!(f, "Given node alias is invalid."),
+			Self::InvalidForwardedPaymentBucketLength => {
+				write!(f, "Forwarded-payment bucket length is invalid.")
+			},
+			Self::ForwardedPaymentBucketLengthMismatch => {
+				write!(f, "Forwarded-payment bucket length does not match persisted data.")
+			},
 			Self::NetworkMismatch => {
 				write!(f, "Given network does not match the node's previously configured network.")
 			},
@@ -1409,6 +1428,12 @@ fn build_with_store_internal(
 ) -> Result<Node, BuildError> {
 	optionally_install_rustls_cryptoprovider();
 
+	let configured_forwarded_payment_bucket_size_secs =
+		config.forwarded_payment_tracking_mode.bucket_size_secs().map_err(|_| {
+			log_error!(logger, "Forwarded-payment bucket length is zero or too large");
+			BuildError::InvalidForwardedPaymentBucketLength
+		})?;
+
 	if let Err(err) = may_announce_channel(&config) {
 		if config.announcement_addresses.is_some() {
 			log_error!(logger, "Announcement addresses were set but some required configuration options for node announcement are missing: {}", err);
@@ -1440,24 +1465,72 @@ 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_all_objects(
-					&*kv_store_ref,
-					PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
-					PAYMENT_INFO_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),
-				)
+	let (
+		payment_store_res,
+		forwarded_payment_store_res,
+		channel_forwarding_stats_res,
+		channel_pair_forwarding_stats_res,
+		forwarded_payment_bucket_size_res,
+		node_metris_res,
+		pending_payment_store_res,
+	) = runtime.block_on(async move {
+		tokio::join!(
+			read_all_objects(
+				&*kv_store_ref,
+				PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
+				PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
+				Arc::clone(&logger_ref),
+			),
+			read_all_objects(
+				&*kv_store_ref,
+				FORWARDED_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
+				FORWARDED_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
+				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_all_objects(
+				&*kv_store_ref,
+				CHANNEL_PAIR_FORWARDING_STATS_PERSISTENCE_PRIMARY_NAMESPACE,
+				CHANNEL_PAIR_FORWARDING_STATS_PERSISTENCE_SECONDARY_NAMESPACE,
+				Arc::clone(&logger_ref),
+			),
+			read_forwarded_payment_bucket_size(&*kv_store_ref, 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),
 			)
-		});
+		)
+	});
+
+	let persisted_forwarded_payment_bucket_size_secs = match forwarded_payment_bucket_size_res {
+		Ok(bucket_size_secs) => Some(bucket_size_secs),
+		Err(e) if e.kind() == std::io::ErrorKind::NotFound => None,
+		Err(e) => {
+			log_error!(logger, "Failed to read forwarded-payment bucket size: {}", e);
+			return Err(BuildError::ReadFailed);
+		},
+	};
+	let forwarded_payment_bucket_config = ResolvedForwardedPaymentBucketConfig::resolve(
+		configured_forwarded_payment_bucket_size_secs,
+		persisted_forwarded_payment_bucket_size_secs,
+	)
+	.map_err(|_| {
+		log_error!(
+			logger,
+			"Configured forwarded-payment bucket size {:?} does not match persisted size {:?}",
+			configured_forwarded_payment_bucket_size_secs,
+			persisted_forwarded_payment_bucket_size_secs
+		);
+		BuildError::ForwardedPaymentBucketLengthMismatch
+	})?;
 
 	// Initialize the status fields.
 	let node_metrics = match node_metris_res {
@@ -1486,6 +1559,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());
@@ -2307,6 +2422,15 @@ fn build_with_store_internal(
 		})
 	});
 
+	if let Some(bucket_size_secs) = forwarded_payment_bucket_config.new_bucket_size_secs() {
+		runtime
+			.block_on(write_forwarded_payment_bucket_size(&*kv_store, bucket_size_secs))
+			.map_err(|e| {
+				log_error!(logger, "Failed to persist forwarded-payment bucket size: {}", e);
+				BuildError::WriteFailed
+			})?;
+	}
+
 	Ok(Node {
 		runtime,
 		stop_sender,
@@ -2334,6 +2458,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_bucket_size_secs: forwarded_payment_bucket_config.bucket_size_secs(),
 		lnurl_auth,
 		is_running,
 		node_metrics,