Add BOLT12 LSPS2 JIT receive support

CHANGELOG.md

@@ -39,6 +39,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

@@ -88,6 +88,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};
@@ -96,6 +97,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,
@@ -2087,12 +2089,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);
@@ -2176,6 +2182,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() {
@@ -2255,7 +2270,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 {},
@@ -2268,7 +2283,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 {},
@@ -2310,6 +2325,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),
 		);
@@ -2395,6 +2411,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),
@@ -2464,12 +2486,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),
@@ -2537,6 +2563,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

@@ -42,6 +42,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::{ChainMonitor, ChannelManager, DynStore, Sweeper, Wallet};
 use crate::{Error, PersistedNodeMetrics};
@@ -181,7 +182,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>,
 	) {
 		// First register for the wallet polling status to make sure `Node::sync_wallets` calls
 		// wait on the result before proceeding.
@@ -210,11 +212,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 mut 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)),
 			];
 
@@ -250,6 +254,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),
 					);
@@ -367,6 +372,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)
 						) => {}
@@ -424,7 +430,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");
@@ -446,6 +453,7 @@ impl BitcoindChainSource {
 			.poll_and_update_listeners_inner(
 				onchain_wallet,
 				channel_manager,
+				offers_message_handler,
 				chain_monitor,
 				output_sweeper,
 			)
@@ -458,7 +466,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() {
@@ -473,6 +482,7 @@ impl BitcoindChainSource {
 				HeaderCache::new(),
 				Arc::clone(&onchain_wallet),
 				Arc::clone(&channel_manager),
+				Arc::clone(&offers_message_handler),
 				chain_monitor,
 				output_sweeper,
 			));
@@ -539,12 +549,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),
 		});
@@ -1537,17 +1549,25 @@ pub(crate) enum FeeRateEstimationMode {
 pub(crate) struct ChainListener {
 	pub(crate) onchain_wallet: std::sync::Weak<Wallet>,
 	pub(crate) channel_manager: std::sync::Weak<ChannelManager>,
+	pub(crate) offers_message_handler: std::sync::Weak<NodeOffersMessageHandler>,
 	pub(crate) chain_monitor: std::sync::Weak<ChainMonitor>,
 	pub(crate) output_sweeper: std::sync::Weak<Sweeper>,
 }
 
 impl ChainListener {
 	fn upgrade(
 		&self,
-	) -> Option<(Arc<Wallet>, Arc<ChannelManager>, Arc<ChainMonitor>, Arc<Sweeper>)> {
+	) -> Option<(
+		Arc<Wallet>,
+		Arc<ChannelManager>,
+		Arc<NodeOffersMessageHandler>,
+		Arc<ChainMonitor>,
+		Arc<Sweeper>,
+	)> {
 		Some((
 			self.onchain_wallet.upgrade()?,
 			self.channel_manager.upgrade()?,
+			self.offers_message_handler.upgrade()?,
 			self.chain_monitor.upgrade()?,
 			self.output_sweeper.upgrade()?,
 		))
@@ -1559,32 +1579,50 @@ impl Listen for ChainListener {
 		&self, header: &bitcoin::block::Header,
 		txdata: &lightning::chain::transaction::TransactionData, height: u32,
 	) {
-		if let Some((onchain_wallet, channel_manager, chain_monitor, output_sweeper)) =
-			self.upgrade()
+		if let Some((
+			onchain_wallet,
+			channel_manager,
+			offers_message_handler,
+			chain_monitor,
+			output_sweeper,
+		)) = self.upgrade()
 		{
 			onchain_wallet.filtered_block_connected(header, txdata, height);
 			channel_manager.filtered_block_connected(header, txdata, height);
+			offers_message_handler.filtered_block_connected(header, txdata, height);
 			chain_monitor.filtered_block_connected(header, txdata, height);
 			output_sweeper.filtered_block_connected(header, txdata, height);
 		}
 	}
 	fn block_connected(&self, block: &bitcoin::Block, height: u32) {
-		if let Some((onchain_wallet, channel_manager, chain_monitor, output_sweeper)) =
-			self.upgrade()
+		if let Some((
+			onchain_wallet,
+			channel_manager,
+			offers_message_handler,
+			chain_monitor,
+			output_sweeper,
+		)) = self.upgrade()
 		{
 			onchain_wallet.block_connected(block, height);
 			channel_manager.block_connected(block, height);
+			offers_message_handler.block_connected(block, height);
 			chain_monitor.block_connected(block, height);
 			output_sweeper.block_connected(block, height);
 		}
 	}
 
 	fn blocks_disconnected(&self, fork_point_block: lightning::chain::BlockLocator) {
-		if let Some((onchain_wallet, channel_manager, chain_monitor, output_sweeper)) =
-			self.upgrade()
+		if let Some((
+			onchain_wallet,
+			channel_manager,
+			offers_message_handler,
+			chain_monitor,
+			output_sweeper,
+		)) = self.upgrade()
 		{
 			onchain_wallet.blocks_disconnected(fork_point_block);
 			channel_manager.blocks_disconnected(fork_point_block);
+			offers_message_handler.blocks_disconnected(fork_point_block);
 			chain_monitor.blocks_disconnected(fork_point_block);
 			output_sweeper.blocks_disconnected(fork_point_block);
 		}

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>,
 		];