Add forwarded payment tracking

Store individual forwarding events and aggregate them into channel and
channel-pair statistics for fee and profitability tracking.

Infer channel-pair allocations for multi-HTLC forwards by matching
incoming and outgoing amounts in FIFO order.

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;
@@ -113,6 +115,7 @@ interface Node {
 	OnchainPayment onchain_payment();
 	UnifiedPayment unified_payment();
 	Liquidity liquidity();
+	Forwarding forwarding();
 	[Throws=NodeError]
 	void lnurl_auth(string lnurl);
 	[Throws=NodeError]
@@ -184,6 +187,8 @@ typedef interface UnifiedPayment;
 
 typedef interface Liquidity;
 
+typedef interface Forwarding;
+
 [Error]
 enum NodeError {
 	"AlreadyRunning",
@@ -395,6 +400,9 @@ typedef string OfferId;
 [Custom]
 typedef string PaymentId;
 
+[Custom]
+typedef string ForwardedPaymentId;
+
 [Custom]
 typedef string PaymentHash;
 
@@ -407,6 +415,12 @@ typedef string PaymentSecret;
 [Custom]
 typedef string ChannelId;
 
+[Custom]
+typedef string ChannelPairStatsId;
+
+[Custom]
+typedef string PageToken;
+
 [Custom]
 typedef string UserChannelId;
 
@@ -433,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

@@ -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.")
 			},
@@ -1408,6 +1427,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);
@@ -1439,24 +1464,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 {
@@ -1485,6 +1558,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());
@@ -2306,6 +2421,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,
@@ -2333,6 +2457,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,