Add BOLT12 LSPS2 JIT receive support

CHANGELOG.md

@@ -43,6 +43,11 @@
   This replaces the previous unpaginated `Node::list_payments`, and
   `Node::list_payments_with_filter` has been removed; filter the returned pages instead.
 - `Node::payment` now returns a `Result`, as retrieving a payment may fail.
+- LSPS2 just-in-time receive fee limits are now configured globally through
+  `Config::lsps2_max_total_lsp_fee_limit_msat`. The corresponding per-call fee-limit arguments
+  have been removed from the BOLT 11 receive APIs.
+- `PaymentKind::Bolt12Offer` now exposes `counterparty_skimmed_fee_msat` for the fee deducted by
+  an LSP when an inbound payment is received through an LSPS2 just-in-time channel.
 - The Bitcoin Core RPC and REST chain-source builder methods now accept an optional
   `wallet_rescan_from_height` argument. Passing a height lets fresh wallets rescan from a known
   birthday block instead of checkpointing at the current tip, which is useful when restoring a

src/builder.rs

@@ -91,6 +91,7 @@ use crate::io::{
 	PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
 	PENDING_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
 };
+use crate::liquidity::client::lsps2::router::LSPS2Router;
 use crate::liquidity::{LSPS2ServiceConfig, LiquiditySourceBuilder, LspConfig};
 use crate::lnurl_auth::LnurlAuth;
 use crate::logger::{log_error, LdkLogger, LogLevel, LogWriter, Logger};
@@ -99,6 +100,7 @@ use crate::payment::asynchronous::om_mailbox::OnionMessageMailbox;
 use crate::payment::forwarding_store::ForwardingStore;
 #[cfg(feature = "unified-payments")]
 use crate::payment::HRNResolver;
+use crate::payment::NodeOffersMessageHandler;
 use crate::peer_store::PeerStore;
 use crate::probing::{
 	HighDegreeStrategy, Prober, ProbingConfig, ProbingStrategy, ProbingStrategyKind,
@@ -2128,12 +2130,16 @@ fn build_with_store_internal(
 	}
 
 	let scoring_fee_params = ProbabilisticScoringFeeParameters::default();
-	let router = Arc::new(DefaultRouter::new(
+	let router = Arc::new(LSPS2Router::new(
+		DefaultRouter::new(
+			Arc::clone(&network_graph),
+			Arc::clone(&logger),
+			Arc::clone(&keys_manager),
+			Arc::clone(&scorer),
+			scoring_fee_params,
+		),
 		Arc::clone(&network_graph),
-		Arc::clone(&logger),
 		Arc::clone(&keys_manager),
-		Arc::clone(&scorer),
-		scoring_fee_params,
 	));
 
 	let mut user_config = default_user_config(&config);
@@ -2217,6 +2223,15 @@ fn build_with_store_internal(
 	};
 
 	let channel_manager = Arc::new(channel_manager);
+	let offers_message_handler = Arc::new(NodeOffersMessageHandler::new(
+		config.network,
+		cur_time.as_secs().try_into().map_err(|_| BuildError::InvalidSystemTime)?,
+		Arc::clone(&channel_manager),
+		Arc::clone(&keys_manager),
+		Arc::clone(&router),
+		Arc::clone(&message_router),
+		Arc::clone(&logger),
+	));
 
 	// Give ChannelMonitors to ChainMonitor
 	for (_blockhash, channel_monitor) in channel_monitors.into_iter() {
@@ -2296,7 +2311,7 @@ fn build_with_store_internal(
 				Arc::clone(&logger),
 				Arc::clone(&channel_manager),
 				message_router,
-				Arc::clone(&channel_manager),
+				Arc::clone(&offers_message_handler),
 				Arc::clone(&channel_manager),
 				Arc::clone(&om_resolver),
 				IgnoringMessageHandler {},
@@ -2309,7 +2324,7 @@ fn build_with_store_internal(
 				Arc::clone(&logger),
 				Arc::clone(&channel_manager),
 				message_router,
-				Arc::clone(&channel_manager),
+				Arc::clone(&offers_message_handler),
 				Arc::clone(&channel_manager),
 				Arc::clone(&om_resolver),
 				IgnoringMessageHandler {},
@@ -2351,6 +2366,7 @@ fn build_with_store_internal(
 			Arc::clone(&keys_manager),
 			Arc::clone(&tx_broadcaster),
 			Arc::clone(&kv_store),
+			Arc::downgrade(&runtime),
 			Arc::clone(&config),
 			Arc::clone(&logger),
 		);
@@ -2436,6 +2452,12 @@ fn build_with_store_internal(
 		Arc::clone(&keys_manager),
 		Arc::clone(&logger),
 	));
+	offers_message_handler.initialize_jit_handling(
+		Arc::clone(&runtime),
+		liquidity_source.lsps2_client(),
+		Arc::downgrade(&connection_manager),
+		Arc::downgrade(&onion_messenger),
+	);
 
 	let output_sweeper = match sweeper_bytes_res {
 		Ok(output_sweeper) => Arc::new(output_sweeper),
@@ -2514,12 +2536,16 @@ fn build_with_store_internal(
 				if let Some(penalty) = probing_cfg.diversity_penalty_msat {
 					probing_fee_params.probing_diversity_penalty_msat = penalty;
 				}
-				let probing_router = Arc::new(DefaultRouter::new(
+				let probing_router = Arc::new(LSPS2Router::new(
+					DefaultRouter::new(
+						Arc::clone(&network_graph),
+						Arc::clone(&logger),
+						Arc::clone(&keys_manager),
+						Arc::clone(&scorer),
+						probing_fee_params,
+					),
 					Arc::clone(&network_graph),
-					Arc::clone(&logger),
 					Arc::clone(&keys_manager),
-					Arc::clone(&scorer),
-					probing_fee_params,
 				));
 				Arc::new(HighDegreeStrategy::new(
 					Arc::clone(&network_graph),
@@ -2587,6 +2613,7 @@ fn build_with_store_internal(
 		output_sweeper,
 		peer_manager,
 		onion_messenger,
+		offers_message_handler,
 		connection_manager,
 		keys_manager,
 		network_graph,

src/chain/bitcoind.rs

@@ -44,6 +44,7 @@ use crate::fee_estimator::{
 };
 use crate::io::utils::update_and_persist_node_metrics;
 use crate::logger::{log_bytes, log_debug, log_error, log_info, log_trace, LdkLogger, Logger};
+use crate::payment::NodeOffersMessageHandler;
 use crate::tx_broadcaster::SortedTransactions;
 use crate::types::{Broadcaster, ChainMonitor, ChannelManager, DynStore, Sweeper, Wallet};
 use crate::{Error, PersistedNodeMetrics};
@@ -257,8 +258,8 @@ impl BitcoindChainSource {
 	pub(super) async fn initialize_wallets(
 		&self, mut stop_sync_receiver: tokio::sync::watch::Receiver<()>,
 		onchain_wallet: Arc<Wallet>, channel_manager: Arc<ChannelManager>,
-		chain_monitor: Arc<ChainMonitor>, output_sweeper: Arc<Sweeper>,
-		tx_broadcaster: &Broadcaster,
+		offers_message_handler: Arc<NodeOffersMessageHandler>, chain_monitor: Arc<ChainMonitor>,
+		output_sweeper: Arc<Sweeper>, tx_broadcaster: &Broadcaster,
 	) -> Result<(), Error> {
 		// First register for the wallet polling status to make sure `Node::sync_wallets` calls
 		// wait on the result before proceeding.
@@ -283,11 +284,13 @@ impl BitcoindChainSource {
 
 		let onchain_wallet_best_block = onchain_wallet.current_best_block();
 		let channel_manager_best_block = channel_manager.current_best_block();
+		let offers_best_block = offers_message_handler.current_best_block();
 		let sweeper_best_block = output_sweeper.current_best_block();
 
 		let chain_listeners = vec![
 			(onchain_wallet_best_block, &*onchain_wallet as &(dyn Listen + Send + Sync)),
 			(channel_manager_best_block, &*channel_manager as &(dyn Listen + Send + Sync)),
+			(offers_best_block, &*offers_message_handler as &(dyn Listen + Send + Sync)),
 			(sweeper_best_block, &*output_sweeper as &(dyn Listen + Send + Sync)),
 		];
 
@@ -309,6 +312,7 @@ impl BitcoindChainSource {
 					header_cache,
 					Arc::clone(&onchain_wallet),
 					Arc::clone(&channel_manager),
+					Arc::clone(&offers_message_handler),
 					Arc::clone(&chain_monitor),
 					Arc::clone(&output_sweeper),
 				);
@@ -350,7 +354,8 @@ impl BitcoindChainSource {
 	pub(super) async fn continuously_sync_wallets(
 		&self, mut stop_sync_receiver: tokio::sync::watch::Receiver<()>,
 		onchain_wallet: Arc<Wallet>, channel_manager: Arc<ChannelManager>,
-		chain_monitor: Arc<ChainMonitor>, output_sweeper: Arc<Sweeper>,
+		offers_message_handler: Arc<NodeOffersMessageHandler>, chain_monitor: Arc<ChainMonitor>,
+		output_sweeper: Arc<Sweeper>,
 	) {
 		let mut chain_polling_interval =
 			tokio::time::interval(Duration::from_secs(CHAIN_POLLING_INTERVAL_SECS));
@@ -389,6 +394,7 @@ impl BitcoindChainSource {
 						_ = self.poll_and_update_listeners(
 							Arc::clone(&onchain_wallet),
 							Arc::clone(&channel_manager),
+							Arc::clone(&offers_message_handler),
 							Arc::clone(&chain_monitor),
 							Arc::clone(&output_sweeper)
 						) => {}
@@ -446,7 +452,8 @@ impl BitcoindChainSource {
 
 	pub(super) async fn poll_and_update_listeners(
 		&self, onchain_wallet: Arc<Wallet>, channel_manager: Arc<ChannelManager>,
-		chain_monitor: Arc<ChainMonitor>, output_sweeper: Arc<Sweeper>,
+		offers_message_handler: Arc<NodeOffersMessageHandler>, chain_monitor: Arc<ChainMonitor>,
+		output_sweeper: Arc<Sweeper>,
 	) -> Result<(), Error> {
 		let receiver_res = {
 			let mut status_lock = self.wallet_polling_status.lock().expect("lock");
@@ -468,6 +475,7 @@ impl BitcoindChainSource {
 			.poll_and_update_listeners_inner(
 				onchain_wallet,
 				channel_manager,
+				offers_message_handler,
 				chain_monitor,
 				output_sweeper,
 			)
@@ -480,7 +488,8 @@ impl BitcoindChainSource {
 
 	async fn poll_and_update_listeners_inner(
 		&self, onchain_wallet: Arc<Wallet>, channel_manager: Arc<ChannelManager>,
-		chain_monitor: Arc<ChainMonitor>, output_sweeper: Arc<Sweeper>,
+		offers_message_handler: Arc<NodeOffersMessageHandler>, chain_monitor: Arc<ChainMonitor>,
+		output_sweeper: Arc<Sweeper>,
 	) -> Result<(), Error> {
 		let mut spv_client_lock = self.spv_client.lock().await;
 		if spv_client_lock.is_none() {
@@ -495,6 +504,7 @@ impl BitcoindChainSource {
 				HeaderCache::new(),
 				Arc::clone(&onchain_wallet),
 				Arc::clone(&channel_manager),
+				Arc::clone(&offers_message_handler),
 				chain_monitor,
 				output_sweeper,
 			));
@@ -561,12 +571,14 @@ impl BitcoindChainSource {
 	fn new_spv_client(
 		&self, chain_tip: ValidatedBlockHeader, header_cache: HeaderCache,
 		onchain_wallet: Arc<Wallet>, channel_manager: Arc<ChannelManager>,
-		chain_monitor: Arc<ChainMonitor>, output_sweeper: Arc<Sweeper>,
+		offers_message_handler: Arc<NodeOffersMessageHandler>, chain_monitor: Arc<ChainMonitor>,
+		output_sweeper: Arc<Sweeper>,
 	) -> BitcoindSpvClient {
 		let chain_poller = ChainPoller::new(Arc::clone(&self.api_client), self.config.network);
 		let chain_listener = Arc::new(ChainListener {
 			onchain_wallet: Arc::downgrade(&onchain_wallet),
 			channel_manager: Arc::downgrade(&channel_manager),
+			offers_message_handler: Arc::downgrade(&offers_message_handler),
 			chain_monitor: Arc::downgrade(&chain_monitor),
 			output_sweeper: Arc::downgrade(&output_sweeper),
 		});

src/chain/electrum.rs

@@ -37,6 +37,7 @@ use crate::fee_estimator::{
 };
 use crate::io::utils::update_and_persist_node_metrics;
 use crate::logger::{log_bytes, log_debug, log_error, log_trace, log_warn, LdkLogger, Logger};
+use crate::payment::NodeOffersMessageHandler;
 use crate::runtime::Runtime;
 use crate::tx_broadcaster::SortedTransactions;
 use crate::types::{ChainMonitor, ChannelManager, DynStore, Sweeper, Wallet};
@@ -220,7 +221,8 @@ impl ElectrumChainSource {
 	}
 
 	pub(crate) async fn sync_lightning_wallet(
-		&self, channel_manager: Arc<ChannelManager>, chain_monitor: Arc<ChainMonitor>,
+		&self, channel_manager: Arc<ChannelManager>,
+		offers_message_handler: Arc<NodeOffersMessageHandler>, chain_monitor: Arc<ChainMonitor>,
 		output_sweeper: Arc<Sweeper>,
 	) -> Result<(), Error> {
 		let receiver_res = {
@@ -238,23 +240,32 @@ impl ElectrumChainSource {
 		let sync_guard =
 			WalletSyncGuard::new(&self.lightning_wallet_sync_status, Error::TxSyncFailed);
 
-		let res =
-			self.sync_lightning_wallet_inner(channel_manager, chain_monitor, output_sweeper).await;
+		let res = self
+			.sync_lightning_wallet_inner(
+				channel_manager,
+				offers_message_handler,
+				chain_monitor,
+				output_sweeper,
+			)
+			.await;
 
 		sync_guard.complete(res);
 
 		res
 	}
 
 	async fn sync_lightning_wallet_inner(
-		&self, channel_manager: Arc<ChannelManager>, chain_monitor: Arc<ChainMonitor>,
+		&self, channel_manager: Arc<ChannelManager>,
+		offers_message_handler: Arc<NodeOffersMessageHandler>, chain_monitor: Arc<ChainMonitor>,
 		output_sweeper: Arc<Sweeper>,
 	) -> Result<(), Error> {
 		let sync_cman = Arc::clone(&channel_manager);
+		let sync_offers = Arc::clone(&offers_message_handler);
 		let sync_cmon = Arc::clone(&chain_monitor);
 		let sync_sweeper = Arc::clone(&output_sweeper);
 		let confirmables: Vec<Arc<dyn Confirm + Sync + Send>> = vec![
 			sync_cman as Arc<dyn Confirm + Sync + Send>,
+			sync_offers as Arc<dyn Confirm + Sync + Send>,
 			sync_cmon as Arc<dyn Confirm + Sync + Send>,
 			sync_sweeper as Arc<dyn Confirm + Sync + Send>,
 		];

src/chain/esplora.rs

@@ -29,6 +29,7 @@ use crate::fee_estimator::{
 };
 use crate::io::utils::update_and_persist_node_metrics;
 use crate::logger::{log_bytes, log_debug, log_error, log_trace, log_warn, LdkLogger, Logger};
+use crate::payment::NodeOffersMessageHandler;
 use crate::tx_broadcaster::SortedTransactions;
 use crate::types::{ChainMonitor, ChannelManager, DynStore, Sweeper, Wallet};
 use crate::{Error, PersistedNodeMetrics};
@@ -270,7 +271,8 @@ impl EsploraChainSource {
 	}
 
 	pub(super) async fn sync_lightning_wallet(
-		&self, channel_manager: Arc<ChannelManager>, chain_monitor: Arc<ChainMonitor>,
+		&self, channel_manager: Arc<ChannelManager>,
+		offers_message_handler: Arc<NodeOffersMessageHandler>, chain_monitor: Arc<ChainMonitor>,
 		output_sweeper: Arc<Sweeper>,
 	) -> Result<(), Error> {
 		let receiver_res = {
@@ -288,23 +290,32 @@ impl EsploraChainSource {
 		let sync_guard =
 			WalletSyncGuard::new(&self.lightning_wallet_sync_status, Error::WalletOperationFailed);
 
-		let res =
-			self.sync_lightning_wallet_inner(channel_manager, chain_monitor, output_sweeper).await;
+		let res = self
+			.sync_lightning_wallet_inner(
+				channel_manager,
+				offers_message_handler,
+				chain_monitor,
+				output_sweeper,
+			)
+			.await;
 
 		sync_guard.complete(res);
 
 		res
 	}
 
 	async fn sync_lightning_wallet_inner(
-		&self, channel_manager: Arc<ChannelManager>, chain_monitor: Arc<ChainMonitor>,
+		&self, channel_manager: Arc<ChannelManager>,
+		offers_message_handler: Arc<NodeOffersMessageHandler>, chain_monitor: Arc<ChainMonitor>,
 		output_sweeper: Arc<Sweeper>,
 	) -> Result<(), Error> {
 		let sync_cman = Arc::clone(&channel_manager);
+		let sync_offers = Arc::clone(&offers_message_handler);
 		let sync_cmon = Arc::clone(&chain_monitor);
 		let sync_sweeper = Arc::clone(&output_sweeper);
 		let confirmables = vec![
 			&*sync_cman as &(dyn Confirm + Sync + Send),
+			&*sync_offers as &(dyn Confirm + Sync + Send),
 			&*sync_cmon as &(dyn Confirm + Sync + Send),
 			&*sync_sweeper as &(dyn Confirm + Sync + Send),
 		];