Add BOLT12 LSPS2 JIT receive support

src/builder.rs

@@ -69,11 +69,13 @@ 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};
 use crate::message_handler::NodeCustomMessageHandler;
 use crate::payment::asynchronous::om_mailbox::OnionMessageMailbox;
+use crate::payment::NodeOffersMessageHandler;
 use crate::peer_store::PeerStore;
 use crate::probing::{
 	HighDegreeStrategy, Prober, ProbingConfig, ProbingStrategy, ProbingStrategyKind,
@@ -1900,12 +1902,15 @@ fn build_with_store_internal(
 	}
 
 	let scoring_fee_params = ProbabilisticScoringFeeParameters::default();
-	let router = Arc::new(DefaultRouter::new(
-		Arc::clone(&network_graph),
-		Arc::clone(&logger),
+	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(&keys_manager),
-		Arc::clone(&scorer),
-		scoring_fee_params,
 	));
 
 	let mut user_config = default_user_config(&config);
@@ -1980,6 +1985,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() {
@@ -2050,7 +2064,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 {},
@@ -2063,7 +2077,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 {},
@@ -2105,6 +2119,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),
 		);
@@ -2189,6 +2204,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),
@@ -2258,12 +2279,15 @@ 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(
-					Arc::clone(&network_graph),
-					Arc::clone(&logger),
+				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(&keys_manager),
-					Arc::clone(&scorer),
-					probing_fee_params,
 				));
 				Arc::new(HighDegreeStrategy::new(
 					Arc::clone(&network_graph),
@@ -2321,6 +2345,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};
@@ -156,7 +157,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.
@@ -184,11 +186,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)),
 			];
 
@@ -224,6 +228,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),
 					);
@@ -341,6 +346,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)
 						) => {}
@@ -398,7 +404,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");
@@ -418,6 +425,7 @@ impl BitcoindChainSource {
 			.poll_and_update_listeners_inner(
 				onchain_wallet,
 				channel_manager,
+				offers_message_handler,
 				chain_monitor,
 				output_sweeper,
 			)
@@ -430,7 +438,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() {
@@ -445,6 +454,7 @@ impl BitcoindChainSource {
 				HeaderCache::new(),
 				Arc::clone(&onchain_wallet),
 				Arc::clone(&channel_manager),
+				Arc::clone(&offers_message_handler),
 				chain_monitor,
 				output_sweeper,
 			));
@@ -511,12 +521,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),
 		});
@@ -1507,17 +1519,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()?,
 		))
@@ -1529,32 +1549,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};
@@ -208,7 +209,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 = {
@@ -224,8 +226,14 @@ impl ElectrumChainSource {
 			})?;
 		}
 
-		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;
 
 		self.lightning_wallet_sync_status
 			.lock()
@@ -236,14 +244,17 @@ impl ElectrumChainSource {
 	}
 
 	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 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>,
 		];