Implement LiquiditySource CMH

We add a `LiquiditySource` object that is configurable via builder
methods, currently allowing to source inbound liquidity from an LSPS2
service.

In the next commit(s) we'll add corresponding API and event handling.

Author: tnull

Cargo.toml

@@ -35,6 +35,7 @@ lightning-persister = { version = "0.0.121" }
 lightning-background-processor = { version = "0.0.121", features = ["futures"] }
 lightning-rapid-gossip-sync = { version = "0.0.121" }
 lightning-transaction-sync = { version = "0.0.121", features = ["esplora-async-https", "time"] }
+lightning-liquidity = { version = "0.1.0-alpha", features = ["std"] }
 
 #lightning = { git = "https://github.com/lightningdevkit/rust-lightning", branch="main", features = ["std"] }
 #lightning-invoice = { git = "https://github.com/lightningdevkit/rust-lightning", branch="main" }
@@ -43,6 +44,7 @@ lightning-transaction-sync = { version = "0.0.121", features = ["esplora-async-h
 #lightning-background-processor = { git = "https://github.com/lightningdevkit/rust-lightning", branch="main", features = ["futures"] }
 #lightning-rapid-gossip-sync = { git = "https://github.com/lightningdevkit/rust-lightning", branch="main" }
 #lightning-transaction-sync = { git = "https://github.com/lightningdevkit/rust-lightning", branch="main", features = ["esplora-async"] }
+#lightning-liquidity = { git = "https://github.com/lightningdevkit/lightning-liquidity", branch="main", features = ["std"] }
 
 #lightning = { path = "../rust-lightning/lightning", features = ["std"] }
 #lightning-invoice = { path = "../rust-lightning/lightning-invoice" }
@@ -51,6 +53,7 @@ lightning-transaction-sync = { version = "0.0.121", features = ["esplora-async-h
 #lightning-background-processor = { path = "../rust-lightning/lightning-background-processor", features = ["futures"] }
 #lightning-rapid-gossip-sync = { path = "../rust-lightning/lightning-rapid-gossip-sync" }
 #lightning-transaction-sync = { path = "../rust-lightning/lightning-transaction-sync", features = ["esplora-async"] }
+#lightning-liquidity = { path = "../lightning-liquidity", features = ["std"] }
 
 bdk = { version = "0.29.0", default-features = false, features = ["std", "async-interface", "use-esplora-async", "sqlite-bundled", "keys-bip39"]}
 

bindings/ldk_node.udl

@@ -28,6 +28,7 @@ interface Builder {
 	void set_esplora_server(string esplora_server_url);
 	void set_gossip_source_p2p();
 	void set_gossip_source_rgs(string rgs_server_url);
+	void set_liquidity_source_lsps2(SocketAddress address, PublicKey node_id, string? token);
 	void set_storage_dir_path(string storage_dir_path);
 	void set_network(Network network);
 	[Throws=BuildError]

src/builder.rs

@@ -3,7 +3,9 @@ use crate::fee_estimator::OnchainFeeEstimator;
 use crate::gossip::GossipSource;
 use crate::io;
 use crate::io::sqlite_store::SqliteStore;
+use crate::liquidity::LiquiditySource;
 use crate::logger::{log_error, FilesystemLogger, Logger};
+use crate::message_handler::NodeCustomMessageHandler;
 use crate::payment_store::PaymentStore;
 use crate::peer_store::PeerStore;
 use crate::sweep::OutputSweeper;
@@ -40,6 +42,9 @@ use lightning_persister::fs_store::FilesystemStore;
 
 use lightning_transaction_sync::EsploraSyncClient;
 
+use lightning_liquidity::lsps2::client::LSPS2ClientConfig;
+use lightning_liquidity::{LiquidityClientConfig, LiquidityManager};
+
 #[cfg(any(vss, vss_test))]
 use crate::io::vss_store::VssStore;
 use bdk::bitcoin::secp256k1::Secp256k1;
@@ -49,6 +54,7 @@ use bdk::template::Bip84;
 
 use bip39::Mnemonic;
 
+use bitcoin::secp256k1::PublicKey;
 use bitcoin::{BlockHash, Network};
 
 #[cfg(any(vss, vss_test))]
@@ -80,6 +86,18 @@ enum GossipSourceConfig {
 	RapidGossipSync(String),
 }
 
+#[derive(Debug, Clone)]
+struct LiquiditySourceConfig {
+	// LSPS2 service's (address, node_id, token)
+	lsps2_service: Option<(SocketAddress, PublicKey, Option<String>)>,
+}
+
+impl Default for LiquiditySourceConfig {
+	fn default() -> Self {
+		Self { lsps2_service: None }
+	}
+}
+
 /// An error encountered during building a [`Node`].
 ///
 /// [`Node`]: crate::Node
@@ -146,16 +164,14 @@ pub struct NodeBuilder {
 	entropy_source_config: Option<EntropySourceConfig>,
 	chain_data_source_config: Option<ChainDataSourceConfig>,
 	gossip_source_config: Option<GossipSourceConfig>,
+	liquidity_source_config: Option<LiquiditySourceConfig>,
 }
 
 impl NodeBuilder {
 	/// Creates a new builder instance with the default configuration.
 	pub fn new() -> Self {
 		let config = Config::default();
-		let entropy_source_config = None;
-		let chain_data_source_config = None;
-		let gossip_source_config = None;
-		Self { config, entropy_source_config, chain_data_source_config, gossip_source_config }
+		Self::from_config(config)
 	}
 
 	/// Creates a new builder instance from an [`Config`].
@@ -164,7 +180,14 @@ impl NodeBuilder {
 		let entropy_source_config = None;
 		let chain_data_source_config = None;
 		let gossip_source_config = None;
-		Self { config, entropy_source_config, chain_data_source_config, gossip_source_config }
+		let liquidity_source_config = None;
+		Self {
+			config,
+			entropy_source_config,
+			chain_data_source_config,
+			gossip_source_config,
+			liquidity_source_config,
+		}
 	}
 
 	/// Configures the [`Node`] instance to source its wallet entropy from a seed file on disk.
@@ -218,6 +241,25 @@ impl NodeBuilder {
 		self
 	}
 
+	/// Configures the [`Node`] instance to source its inbound liquidity from the given
+	/// [LSPS2](https://github.com/BitcoinAndLightningLayerSpecs/lsp/blob/main/LSPS2/README.md)
+	/// service.
+	///
+	/// Will mark the LSP as trusted for 0-confirmation channels, see [`Config::trusted_peers_0conf`].
+	///
+	/// The given `token` will be used by the LSP to authenticate the user.
+	pub fn set_liquidity_source_lsps2(
+		&mut self, address: SocketAddress, node_id: PublicKey, token: Option<String>,
+	) -> &mut Self {
+		// Mark the LSP as trusted for 0conf
+		self.config.trusted_peers_0conf.push(node_id.clone());
+
+		let liquidity_source_config =
+			self.liquidity_source_config.get_or_insert(LiquiditySourceConfig::default());
+		liquidity_source_config.lsps2_service = Some((address, node_id, token));
+		self
+	}
+
 	/// Sets the used storage directory path.
 	pub fn set_storage_dir_path(&mut self, storage_dir_path: String) -> &mut Self {
 		self.config.storage_dir_path = storage_dir_path;
@@ -318,6 +360,7 @@ impl NodeBuilder {
 			config,
 			self.chain_data_source_config.as_ref(),
 			self.gossip_source_config.as_ref(),
+			self.liquidity_source_config.as_ref(),
 			seed_bytes,
 			logger,
 			vss_store,
@@ -340,6 +383,7 @@ impl NodeBuilder {
 			config,
 			self.chain_data_source_config.as_ref(),
 			self.gossip_source_config.as_ref(),
+			self.liquidity_source_config.as_ref(),
 			seed_bytes,
 			logger,
 			kv_store,
@@ -413,6 +457,19 @@ impl ArcedNodeBuilder {
 		self.inner.write().unwrap().set_gossip_source_rgs(rgs_server_url);
 	}
 
+	/// Configures the [`Node`] instance to source its inbound liquidity from the given
+	/// [LSPS2](https://github.com/BitcoinAndLightningLayerSpecs/lsp/blob/main/LSPS2/README.md)
+	/// service.
+	///
+	/// Will mark the LSP as trusted for 0-confirmation channels, see [`Config::trusted_peers_0conf`].
+	///
+	/// The given `token` will be used by the LSP to authenticate the user.
+	pub fn set_liquidity_source_lsps2(
+		&self, address: SocketAddress, node_id: PublicKey, token: Option<String>,
+	) {
+		self.inner.write().unwrap().set_liquidity_source_lsps2(address, node_id, token);
+	}
+
 	/// Sets the used storage directory path.
 	pub fn set_storage_dir_path(&self, storage_dir_path: String) {
 		self.inner.write().unwrap().set_storage_dir_path(storage_dir_path);
@@ -463,7 +520,8 @@ impl ArcedNodeBuilder {
 /// Builds a [`Node`] instance according to the options previously configured.
 fn build_with_store_internal<K: KVStore + Sync + Send + 'static>(
 	config: Arc<Config>, chain_data_source_config: Option<&ChainDataSourceConfig>,
-	gossip_source_config: Option<&GossipSourceConfig>, seed_bytes: [u8; 64],
+	gossip_source_config: Option<&GossipSourceConfig>,
+	liquidity_source_config: Option<&LiquiditySourceConfig>, seed_bytes: [u8; 64],
 	logger: Arc<FilesystemLogger>, kv_store: Arc<K>,
 ) -> Result<Node<K>, BuildError> {
 	// Initialize the on-chain wallet and chain access
@@ -746,20 +804,51 @@ fn build_with_store_internal<K: KVStore + Sync + Send + 'static>(
 		}
 	};
 
+	let liquidity_source = liquidity_source_config.as_ref().and_then(|lsc| {
+		lsc.lsps2_service.as_ref().map(|(address, node_id, token)| {
+			let lsps2_client_config = Some(LSPS2ClientConfig {});
+			let liquidity_client_config = Some(LiquidityClientConfig { lsps2_client_config });
+			let liquidity_manager = Arc::new(LiquidityManager::new(
+				Arc::clone(&keys_manager),
+				Arc::clone(&channel_manager),
+				Some(Arc::clone(&tx_sync)),
+				None,
+				None,
+				liquidity_client_config,
+			));
+			Arc::new(LiquiditySource::new_lsps2(
+				address.clone(),
+				*node_id,
+				token.clone(),
+				Arc::clone(&channel_manager),
+				Arc::clone(&keys_manager),
+				liquidity_manager,
+				Arc::clone(&config),
+				Arc::clone(&logger),
+			))
+		})
+	});
+
+	let custom_message_handler = if let Some(liquidity_source) = liquidity_source.as_ref() {
+		Arc::new(NodeCustomMessageHandler::new_liquidity(Arc::clone(&liquidity_source)))
+	} else {
+		Arc::new(NodeCustomMessageHandler::new_ignoring())
+	};
+
 	let msg_handler = match gossip_source.as_gossip_sync() {
 		GossipSync::P2P(p2p_gossip_sync) => MessageHandler {
 			chan_handler: Arc::clone(&channel_manager),
 			route_handler: Arc::clone(&p2p_gossip_sync)
 				as Arc<dyn RoutingMessageHandler + Sync + Send>,
 			onion_message_handler: onion_messenger,
-			custom_message_handler: IgnoringMessageHandler {},
+			custom_message_handler,
 		},
 		GossipSync::Rapid(_) => MessageHandler {
 			chan_handler: Arc::clone(&channel_manager),
 			route_handler: Arc::new(IgnoringMessageHandler {})
 				as Arc<dyn RoutingMessageHandler + Sync + Send>,
 			onion_message_handler: onion_messenger,
-			custom_message_handler: IgnoringMessageHandler {},
+			custom_message_handler,
 		},
 		GossipSync::None => {
 			unreachable!("We must always have a gossip sync!");
@@ -782,6 +871,8 @@ fn build_with_store_internal<K: KVStore + Sync + Send + 'static>(
 		Arc::clone(&keys_manager),
 	));
 
+	liquidity_source.as_ref().map(|l| l.set_peer_manager(Arc::clone(&peer_manager)));
+
 	// Init payment info storage
 	let payment_store = match io::utils::read_payments(Arc::clone(&kv_store), Arc::clone(&logger)) {
 		Ok(payments) => {
@@ -853,6 +944,7 @@ fn build_with_store_internal<K: KVStore + Sync + Send + 'static>(
 		keys_manager,
 		network_graph,
 		gossip_source,
+		liquidity_source,
 		kv_store,
 		logger,
 		_router: router,

src/lib.rs

@@ -82,7 +82,9 @@ mod fee_estimator;
 mod gossip;
 mod hex_utils;
 pub mod io;
+mod liquidity;
 mod logger;
+mod message_handler;
 mod payment_store;
 mod peer_store;
 mod sweep;
@@ -116,6 +118,7 @@ pub use builder::NodeBuilder as Builder;
 
 use event::{EventHandler, EventQueue};
 use gossip::GossipSource;
+use liquidity::LiquiditySource;
 use payment_store::PaymentStore;
 pub use payment_store::{PaymentDetails, PaymentDirection, PaymentStatus};
 use peer_store::{PeerInfo, PeerStore};
@@ -311,6 +314,7 @@ pub struct Node<K: KVStore + Sync + Send + 'static> {
 	keys_manager: Arc<KeysManager>,
 	network_graph: Arc<NetworkGraph>,
 	gossip_source: Arc<GossipSource>,
+	liquidity_source: Option<Arc<LiquiditySource<K, Arc<FilesystemLogger>>>>,
 	kv_store: Arc<K>,
 	logger: Arc<FilesystemLogger>,
 	_router: Arc<Router>,
@@ -764,6 +768,21 @@ impl<K: KVStore + Sync + Send + 'static> Node<K> {
 			});
 		});
 
+		if let Some(liquidity_source) = self.liquidity_source.as_ref() {
+			let mut stop_liquidity_handler = self.stop_receiver.clone();
+			let liquidity_handler = Arc::clone(&liquidity_source);
+			runtime.spawn(async move {
+				loop {
+					tokio::select! {
+						_ = stop_liquidity_handler.changed() => {
+							return;
+						}
+						_ = liquidity_handler.handle_next_event() => {}
+					}
+				}
+			});
+		}
+
 		*runtime_lock = Some(runtime);
 
 		log_info!(self.logger, "Startup complete.");

src/liquidity.rs

@@ -0,0 +1,73 @@
+use crate::types::{ChannelManager, KeysManager, LiquidityManager, PeerManager};
+use crate::Config;
+
+use lightning::ln::msgs::SocketAddress;
+use lightning::util::logger::Logger;
+use lightning::util::persist::KVStore;
+use lightning_liquidity::events::Event;
+use lightning_liquidity::lsps2::event::LSPS2ClientEvent;
+
+use bitcoin::secp256k1::PublicKey;
+
+use std::ops::Deref;
+use std::sync::Arc;
+
+struct LSPS2Service {
+	address: SocketAddress,
+	node_id: PublicKey,
+	token: Option<String>,
+}
+
+pub(crate) struct LiquiditySource<K: KVStore + Sync + Send + 'static, L: Deref>
+where
+	L::Target: Logger,
+{
+	lsps2_service: Option<LSPS2Service>,
+	channel_manager: Arc<ChannelManager<K>>,
+	keys_manager: Arc<KeysManager>,
+	liquidity_manager: Arc<LiquidityManager<K>>,
+	config: Arc<Config>,
+	logger: L,
+}
+
+impl<K: KVStore + Sync + Send, L: Deref> LiquiditySource<K, L>
+where
+	L::Target: Logger,
+{
+	pub(crate) fn new_lsps2(
+		address: SocketAddress, node_id: PublicKey, token: Option<String>,
+		channel_manager: Arc<ChannelManager<K>>, keys_manager: Arc<KeysManager>,
+		liquidity_manager: Arc<LiquidityManager<K>>, config: Arc<Config>, logger: L,
+	) -> Self {
+		let lsps2_service = Some(LSPS2Service { address, node_id, token });
+		Self { lsps2_service, channel_manager, keys_manager, liquidity_manager, config, logger }
+	}
+
+	pub(crate) fn set_peer_manager(&self, peer_manager: Arc<PeerManager<K>>) {
+		let process_msgs_callback = move || peer_manager.process_events();
+		self.liquidity_manager.set_process_msgs_callback(process_msgs_callback);
+	}
+
+	pub(crate) fn liquidity_manager(&self) -> &LiquidityManager<K> {
+		self.liquidity_manager.as_ref()
+	}
+
+	pub(crate) async fn handle_next_event(&self) {
+		match self.liquidity_manager.next_event_async().await {
+			Event::LSPS2Client(LSPS2ClientEvent::OpeningParametersReady {
+				counterparty_node_id: _,
+				opening_fee_params_menu: _,
+				min_payment_size_msat: _,
+				max_payment_size_msat: _,
+			}) => {}
+			Event::LSPS2Client(LSPS2ClientEvent::InvoiceParametersReady {
+				counterparty_node_id: _,
+				intercept_scid: _,
+				cltv_expiry_delta: _,
+				payment_size_msat: _,
+				user_channel_id: _,
+			}) => {}
+			_ => {}
+		}
+	}
+}