Add storage for forwarded payments

Routing nodes and LSPs want to track forwarded payments so they can run accounting on fees earned and track profitability across time. We now store these to make it easier to track and allows for future accounting utils in the future.

This shouldn't effect edge user nodes as they should never be forwarding payments.

Implementation is mostly just copied how we currently handle normal payments and adapted for forwarded payments.

Author: benthecarman

bindings/ldk_node.udl

@@ -179,6 +179,8 @@ interface Node {
 	void remove_payment([ByRef]PaymentId payment_id);
 	BalanceDetails list_balances();
 	sequence<PaymentDetails> list_payments();
+	ForwardedPaymentDetails? forwarded_payment([ByRef]ForwardedPaymentId forwarded_payment_id);
+	sequence<ForwardedPaymentDetails> list_forwarded_payments();
 	sequence<PeerDetails> list_peers();
 	sequence<ChannelDetails> list_channels();
 	NetworkGraph network_graph();
@@ -497,6 +499,21 @@ dictionary PaymentDetails {
 	u64 latest_update_timestamp;
 };
 
+dictionary ForwardedPaymentDetails {
+	ForwardedPaymentId id;
+	ChannelId prev_channel_id;
+	ChannelId next_channel_id;
+	UserChannelId? prev_user_channel_id;
+	UserChannelId? next_user_channel_id;
+	PublicKey? prev_node_id;
+	PublicKey? next_node_id;
+	u64? total_fee_earned_msat;
+	u64? skimmed_fee_msat;
+	boolean claim_from_onchain_tx;
+	u64? outbound_amount_forwarded_msat;
+	u64 forwarded_at_timestamp;
+};
+
 dictionary RouteParametersConfig {
 	u64? max_total_routing_fee_msat;
 	u32 max_total_cltv_expiry_delta;
@@ -884,6 +901,9 @@ typedef string OfferId;
 [Custom]
 typedef string PaymentId;
 
+[Custom]
+typedef string ForwardedPaymentId;
+
 [Custom]
 typedef string PaymentHash;
 

src/builder.rs

@@ -54,13 +54,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_network_graph,
-	read_node_metrics, read_output_sweeper, read_payments, read_peer_info, read_pending_payments,
-	read_scorer, write_node_metrics,
+	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, PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
+	self, 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,
 };
@@ -74,9 +76,9 @@ use crate::peer_store::PeerStore;
 use crate::runtime::{Runtime, RuntimeSpawner};
 use crate::tx_broadcaster::TransactionBroadcaster;
 use crate::types::{
-	AsyncPersister, ChainMonitor, ChannelManager, DynStore, DynStoreWrapper, GossipSync, Graph,
-	KeysManager, MessageRouter, OnionMessenger, PaymentStore, PeerManager, PendingPaymentStore,
-	Persister, SyncAndAsyncKVStore,
+	AsyncPersister, ChainMonitor, ChannelManager, DynStore, DynStoreWrapper, ForwardedPaymentStore,
+	GossipSync, Graph, KeysManager, MessageRouter, OnionMessenger, PaymentStore, PeerManager,
+	PendingPaymentStore, Persister, SyncAndAsyncKVStore,
 };
 use crate::wallet::persist::KVStoreWalletPersister;
 use crate::wallet::Wallet;
@@ -1059,14 +1061,19 @@ 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_payments(&*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))
-			)
-		});
+	let (
+		payment_store_res,
+		forwarded_payment_store_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_node_metrics(&*kv_store_ref, Arc::clone(&logger_ref)),
+			read_pending_payments(&*kv_store_ref, Arc::clone(&logger_ref))
+		)
+	});
 
 	// Initialize the status fields.
 	let node_metrics = match node_metris_res {
@@ -1095,6 +1102,20 @@ 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 (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());
@@ -1781,6 +1802,7 @@ fn build_with_store_internal(
 		scorer,
 		peer_store,
 		payment_store,
+		forwarded_payment_store,
 		is_running,
 		node_metrics,
 		om_mailbox,

src/event.rs

@@ -10,6 +10,7 @@ use core::task::{Poll, Waker};
 use std::collections::VecDeque;
 use std::ops::Deref;
 use std::sync::{Arc, Mutex};
+use std::time::{SystemTime, UNIX_EPOCH};
 
 use bitcoin::blockdata::locktime::absolute::LockTime;
 use bitcoin::secp256k1::PublicKey;
@@ -45,10 +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::{
-	PaymentDetails, PaymentDetailsUpdate, PaymentDirection, PaymentKind, PaymentStatus,
+	ForwardedPaymentDetails, ForwardedPaymentId, PaymentDetails, PaymentDetailsUpdate,
+	PaymentDirection, PaymentKind, PaymentStatus,
 };
 use crate::runtime::Runtime;
-use crate::types::{CustomTlvRecord, DynStore, OnionMessenger, PaymentStore, Sweeper, Wallet};
+use crate::types::{
+	CustomTlvRecord, DynStore, ForwardedPaymentStore, OnionMessenger, PaymentStore, Sweeper, Wallet,
+};
 use crate::{
 	hex_utils, BumpTransactionEventHandler, ChannelManager, Error, Graph, PeerInfo, PeerStore,
 	UserChannelId,
@@ -487,6 +491,7 @@ where
 	network_graph: Arc<Graph>,
 	liquidity_source: Option<Arc<LiquiditySource<Arc<Logger>>>>,
 	payment_store: Arc<PaymentStore>,
+	forwarded_payment_store: Arc<ForwardedPaymentStore>,
 	peer_store: Arc<PeerStore<L>>,
 	runtime: Arc<Runtime>,
 	logger: L,
@@ -506,10 +511,10 @@ where
 		channel_manager: Arc<ChannelManager>, connection_manager: Arc<ConnectionManager<L>>,
 		output_sweeper: Arc<Sweeper>, network_graph: Arc<Graph>,
 		liquidity_source: Option<Arc<LiquiditySource<Arc<Logger>>>>,
-		payment_store: Arc<PaymentStore>, 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>,
+		payment_store: Arc<PaymentStore>, forwarded_payment_store: Arc<ForwardedPaymentStore>,
+		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>,
 	) -> Self {
 		Self {
 			event_queue,
@@ -521,6 +526,7 @@ where
 			network_graph,
 			liquidity_source,
 			payment_store,
+			forwarded_payment_store,
 			peer_store,
 			logger,
 			runtime,
@@ -1364,9 +1370,44 @@ 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,
+				};
+
+				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.expect("prev_channel_id expected for events generated by LDK versions greater than 0.0.107."),
-					next_channel_id: next_channel_id.expect("next_channel_id expected for events generated by LDK versions greater than 0.0.107."),
+					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,

src/ffi/types.rs

@@ -54,9 +54,10 @@ pub use crate::graph::{ChannelInfo, ChannelUpdateInfo, NodeAnnouncementInfo, Nod
 pub use crate::liquidity::{LSPS1OrderStatus, LSPS2ServiceConfig};
 pub use crate::logger::{LogLevel, LogRecord, LogWriter};
 pub use crate::payment::store::{
-	ConfirmationStatus, LSPFeeLimits, PaymentDirection, PaymentKind, PaymentStatus,
+	ConfirmationStatus, ForwardedPaymentId, LSPFeeLimits, PaymentDirection, PaymentKind,
+	PaymentStatus,
 };
-pub use crate::payment::UnifiedPaymentResult;
+pub use crate::payment::{ForwardedPaymentDetails, UnifiedPaymentResult};
 use crate::{hex_utils, SocketAddress, UniffiCustomTypeConverter, UserChannelId};
 
 impl UniffiCustomTypeConverter for PublicKey {
@@ -722,6 +723,24 @@ impl UniffiCustomTypeConverter for PaymentId {
 	}
 }
 
+impl UniffiCustomTypeConverter for ForwardedPaymentId {
+	type Builtin = String;
+
+	fn into_custom(val: Self::Builtin) -> uniffi::Result<Self> {
+		if let Some(bytes_vec) = hex_utils::to_vec(&val) {
+			let bytes_res = bytes_vec.try_into();
+			if let Ok(bytes) = bytes_res {
+				return Ok(ForwardedPaymentId(bytes));
+			}
+		}
+		Err(Error::InvalidPaymentId.into())
+	}
+
+	fn from_custom(obj: Self) -> Self::Builtin {
+		hex_utils::to_string(&obj.0)
+	}
+}
+
 impl UniffiCustomTypeConverter for PaymentHash {
 	type Builtin = String;
 

src/io/mod.rs

@@ -27,6 +27,10 @@ pub(crate) const PEER_INFO_PERSISTENCE_KEY: &str = "peers";
 pub(crate) const PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE: &str = "payments";
 pub(crate) const PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE: &str = "";
 
+/// The forwarded payment information will be persisted under this prefix.
+pub(crate) const FORWARDED_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE: &str = "forwarded_payments";
+pub(crate) const FORWARDED_PAYMENT_INFO_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

@@ -46,7 +46,7 @@ use crate::io::{
 	NODE_METRICS_KEY, NODE_METRICS_PRIMARY_NAMESPACE, NODE_METRICS_SECONDARY_NAMESPACE,
 };
 use crate::logger::{log_error, LdkLogger, Logger};
-use crate::payment::PendingPaymentDetails;
+use crate::payment::{ForwardedPaymentDetails, PendingPaymentDetails};
 use crate::peer_store::PeerStore;
 use crate::types::{Broadcaster, DynStore, KeysManager, Sweeper};
 use crate::wallet::ser::{ChangeSetDeserWrapper, ChangeSetSerWrapper};
@@ -304,6 +304,22 @@ where
 	.await
 }
 
+/// Read previously persisted forwarded payments information from the store.
+pub(crate) async fn read_forwarded_payments<L: Deref>(
+	kv_store: &DynStore, logger: L,
+) -> Result<Vec<ForwardedPaymentDetails>, std::io::Error>
+where
+	L::Target: LdkLogger,
+{
+	read_objects_from_store(
+		kv_store,
+		logger,
+		FORWARDED_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
+		FORWARDED_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
+	)
+	.await
+}
+
 /// Read `OutputSweeper` state from the store.
 pub(crate) async fn read_output_sweeper(
 	broadcaster: Arc<Broadcaster>, fee_estimator: Arc<OnchainFeeEstimator>,

src/lib.rs

@@ -153,16 +153,16 @@ use logger::{log_debug, log_error, log_info, log_trace, LdkLogger, Logger};
 use payment::asynchronous::om_mailbox::OnionMessageMailbox;
 use payment::asynchronous::static_invoice_store::StaticInvoiceStore;
 use payment::{
-	Bolt11Payment, Bolt12Payment, OnchainPayment, PaymentDetails, SpontaneousPayment,
-	UnifiedPayment,
+	Bolt11Payment, Bolt12Payment, ForwardedPaymentDetails, ForwardedPaymentId, OnchainPayment,
+	PaymentDetails, SpontaneousPayment, UnifiedPayment,
 };
 use peer_store::{PeerInfo, PeerStore};
 use rand::Rng;
 use runtime::Runtime;
 use types::{
-	Broadcaster, BumpTransactionEventHandler, ChainMonitor, ChannelManager, DynStore, Graph,
-	HRNResolver, KeysManager, OnionMessenger, PaymentStore, PeerManager, Router, Scorer, Sweeper,
-	Wallet,
+	Broadcaster, BumpTransactionEventHandler, ChainMonitor, ChannelManager, DynStore,
+	ForwardedPaymentStore, Graph, HRNResolver, KeysManager, OnionMessenger, PaymentStore,
+	PeerManager, Router, Scorer, Sweeper, Wallet,
 };
 pub use types::{ChannelDetails, CustomTlvRecord, PeerDetails, SyncAndAsyncKVStore, UserChannelId};
 pub use {
@@ -222,6 +222,7 @@ pub struct Node {
 	scorer: Arc<Mutex<Scorer>>,
 	peer_store: Arc<PeerStore<Arc<Logger>>>,
 	payment_store: Arc<PaymentStore>,
+	forwarded_payment_store: Arc<ForwardedPaymentStore>,
 	is_running: Arc<RwLock<bool>>,
 	node_metrics: Arc<RwLock<NodeMetrics>>,
 	om_mailbox: Option<Arc<OnionMessageMailbox>>,
@@ -573,6 +574,7 @@ impl Node {
 			Arc::clone(&self.network_graph),
 			self.liquidity_source.clone(),
 			Arc::clone(&self.payment_store),
+			Arc::clone(&self.forwarded_payment_store),
 			Arc::clone(&self.peer_store),
 			static_invoice_store,
 			Arc::clone(&self.onion_messenger),
@@ -1692,6 +1694,34 @@ impl Node {
 		self.payment_store.list_filter(|_| true)
 	}
 
+	/// Retrieve the details of a specific forwarded payment with the given id.
+	///
+	/// Returns `Some` if the forwarded payment was known and `None` otherwise.
+	pub fn forwarded_payment(
+		&self, forwarded_payment_id: &ForwardedPaymentId,
+	) -> Option<ForwardedPaymentDetails> {
+		self.forwarded_payment_store.get(forwarded_payment_id)
+	}
+
+	/// Retrieves all forwarded payments that match the given predicate.
+	///
+	/// For example, to list all forwarded payments that earned at least 1000 msat in fees:
+	/// ```ignore
+	/// node.list_forwarded_payments_with_filter(|p| {
+	///     p.total_fee_earned_msat.unwrap_or(0) >= 1000
+	/// });
+	/// ```
+	pub fn list_forwarded_payments_with_filter<F: FnMut(&&ForwardedPaymentDetails) -> bool>(
+		&self, f: F,
+	) -> Vec<ForwardedPaymentDetails> {
+		self.forwarded_payment_store.list_filter(f)
+	}
+
+	/// Retrieves all forwarded payments.
+	pub fn list_forwarded_payments(&self) -> Vec<ForwardedPaymentDetails> {
+		self.forwarded_payment_store.list_filter(|_| true)
+	}
+
 	/// Retrieves a list of known peers.
 	pub fn list_peers(&self) -> Vec<PeerDetails> {
 		let mut peers = Vec::new();

src/payment/mod.rs

@@ -22,6 +22,7 @@ pub use onchain::OnchainPayment;
 pub use pending_payment_store::PendingPaymentDetails;
 pub use spontaneous::SpontaneousPayment;
 pub use store::{
-	ConfirmationStatus, LSPFeeLimits, PaymentDetails, PaymentDirection, PaymentKind, PaymentStatus,
+	ConfirmationStatus, ForwardedPaymentDetails, ForwardedPaymentId, LSPFeeLimits, PaymentDetails,
+	PaymentDirection, PaymentKind, PaymentStatus,
 };
 pub use unified::{UnifiedPayment, UnifiedPaymentResult};