Merge pull request #365 from tnull/2024-10-chain-source-refactor

Refactor syncing and introduce `ChainSource`

Author: tnull

README.md

@@ -24,7 +24,7 @@ use std::str::FromStr;
 fn main() {
 	let mut builder = Builder::new();
 	builder.set_network(Network::Testnet);
-	builder.set_esplora_server("https://blockstream.info/testnet/api".to_string());
+	builder.set_chain_source_esplora("https://blockstream.info/testnet/api".to_string(), None);
 	builder.set_gossip_source_rgs("https://rapidsync.lightningdevkit.org/testnet/snapshot".to_string());
 
 	let node = builder.build().unwrap();

bindings/kotlin/ldk-node-jvm/lib/src/test/kotlin/org/lightningdevkit/ldknode/LibraryTest.kt

@@ -130,9 +130,9 @@ class LibraryTest {
         println("Config 2: $config2")
 
         val builder1 = Builder.fromConfig(config1)
-        builder1.setEsploraServer(esploraEndpoint)
+        builder1.setChainSourceEsplora(esploraEndpoint, null)
         val builder2 = Builder.fromConfig(config2)
-        builder2.setEsploraServer(esploraEndpoint)
+        builder2.setChainSourceEsplora(esploraEndpoint, null)
 
         val node1 = builder1.build()
         val node2 = builder2.build()

bindings/ldk_node.udl

@@ -9,9 +9,6 @@ dictionary Config {
 	Network network;
 	sequence<SocketAddress>? listening_addresses;
 	NodeAlias? node_alias;
-	u64 onchain_wallet_sync_interval_secs;
-	u64 wallet_sync_interval_secs;
-	u64 fee_rate_cache_update_interval_secs;
 	sequence<PublicKey> trusted_peers_0conf;
 	u64 probing_liquidity_limit_multiplier;
 	LogLevel log_level;
@@ -24,6 +21,12 @@ dictionary AnchorChannelsConfig {
 	u64 per_channel_reserve_sats;
 };
 
+dictionary EsploraSyncConfig {
+	u64 onchain_wallet_sync_interval_secs;
+	u64 lightning_wallet_sync_interval_secs;
+	u64 fee_rate_cache_update_interval_secs;
+};
+
 interface Builder {
 	constructor();
 	[Name=from_config]
@@ -32,7 +35,7 @@ interface Builder {
 	[Throws=BuildError]
 	void set_entropy_seed_bytes(sequence<u8> seed_bytes);
 	void set_entropy_bip39_mnemonic(Mnemonic mnemonic, string? passphrase);
-	void set_esplora_server(string esplora_server_url);
+	void set_chain_source_esplora(string server_url, EsploraSyncConfig? config);
 	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);
@@ -218,11 +221,12 @@ dictionary NodeStatus {
 	boolean is_running;
 	boolean is_listening;
 	BestBlock current_best_block;
-	u64? latest_wallet_sync_timestamp;
+	u64? latest_lightning_wallet_sync_timestamp;
 	u64? latest_onchain_wallet_sync_timestamp;
 	u64? latest_fee_rate_cache_update_timestamp;
 	u64? latest_rgs_snapshot_timestamp;
 	u64? latest_node_announcement_broadcast_timestamp;
+	u32? latest_channel_monitor_archival_height;
 };
 
 dictionary BestBlock {

bindings/python/src/ldk_node/test_ldk_node.py

@@ -84,7 +84,7 @@ def setup_node(tmp_dir, esplora_endpoint, listening_addresses):
     config = default_config()
     builder = Builder.from_config(config)
     builder.set_storage_dir_path(tmp_dir)
-    builder.set_esplora_server(esplora_endpoint)
+    builder.set_chain_source_esplora(esplora_endpoint, None)
     builder.set_network(DEFAULT_TEST_NETWORK)
     builder.set_listening_addresses(listening_addresses)
     return builder.build()

src/builder.rs

@@ -5,16 +5,15 @@
 // http://opensource.org/licenses/MIT>, at your option. You may not use this file except in
 // accordance with one or both of these licenses.
 
-use crate::config::{
-	default_user_config, Config, DEFAULT_ESPLORA_CLIENT_TIMEOUT_SECS, DEFAULT_ESPLORA_SERVER_URL,
-	WALLET_KEYS_SEED_LEN,
-};
+use crate::chain::{ChainSource, DEFAULT_ESPLORA_SERVER_URL};
+use crate::config::{default_user_config, Config, EsploraSyncConfig, WALLET_KEYS_SEED_LEN};
+
 use crate::connection::ConnectionManager;
 use crate::event::EventQueue;
 use crate::fee_estimator::OnchainFeeEstimator;
 use crate::gossip::GossipSource;
-use crate::io;
 use crate::io::sqlite_store::SqliteStore;
+use crate::io::utils::{read_node_metrics, write_node_metrics};
 #[cfg(any(vss, vss_test))]
 use crate::io::vss_store::VssStore;
 use crate::liquidity::LiquiditySource;
@@ -29,6 +28,7 @@ use crate::types::{
 };
 use crate::wallet::persist::KVStoreWalletPersister;
 use crate::wallet::Wallet;
+use crate::{io, NodeMetrics};
 use crate::{LogLevel, Node};
 
 use lightning::chain::{chainmonitor, BestBlock, Watch};
@@ -52,8 +52,6 @@ use lightning::util::sweep::OutputSweeper;
 
 use lightning_persister::fs_store::FilesystemStore;
 
-use lightning_transaction_sync::EsploraSyncClient;
-
 use lightning_liquidity::lsps2::client::LSPS2ClientConfig;
 use lightning_liquidity::{LiquidityClientConfig, LiquidityManager};
 
@@ -79,7 +77,7 @@ use std::time::SystemTime;
 
 #[derive(Debug, Clone)]
 enum ChainDataSourceConfig {
-	Esplora(String),
+	Esplora { server_url: String, sync_config: Option<EsploraSyncConfig> },
 }
 
 #[derive(Debug, Clone)]
@@ -239,8 +237,14 @@ impl NodeBuilder {
 	}
 
 	/// Configures the [`Node`] instance to source its chain data from the given Esplora server.
-	pub fn set_esplora_server(&mut self, esplora_server_url: String) -> &mut Self {
-		self.chain_data_source_config = Some(ChainDataSourceConfig::Esplora(esplora_server_url));
+	///
+	/// If no `sync_config` is given, default values are used. See [`EsploraSyncConfig`] for more
+	/// information.
+	pub fn set_chain_source_esplora(
+		&mut self, server_url: String, sync_config: Option<EsploraSyncConfig>,
+	) -> &mut Self {
+		self.chain_data_source_config =
+			Some(ChainDataSourceConfig::Esplora { server_url, sync_config });
 		self
 	}
 
@@ -466,8 +470,13 @@ impl ArcedNodeBuilder {
 	}
 
 	/// Configures the [`Node`] instance to source its chain data from the given Esplora server.
-	pub fn set_esplora_server(&self, esplora_server_url: String) {
-		self.inner.write().unwrap().set_esplora_server(esplora_server_url);
+	///
+	/// If no `sync_config` is given, default values are used. See [`EsploraSyncConfig`] for more
+	/// information.
+	pub fn set_chain_source_esplora(
+		&self, server_url: String, sync_config: Option<EsploraSyncConfig>,
+	) {
+		self.inner.write().unwrap().set_chain_source_esplora(server_url, sync_config);
 	}
 
 	/// Configures the [`Node`] instance to source its gossip data from the Lightning peer-to-peer
@@ -555,6 +564,19 @@ fn build_with_store_internal(
 	liquidity_source_config: Option<&LiquiditySourceConfig>, seed_bytes: [u8; 64],
 	logger: Arc<FilesystemLogger>, kv_store: Arc<DynStore>,
 ) -> Result<Node, BuildError> {
+	// Initialize the status fields.
+	let is_listening = Arc::new(AtomicBool::new(false));
+	let node_metrics = match read_node_metrics(Arc::clone(&kv_store), Arc::clone(&logger)) {
+		Ok(metrics) => Arc::new(RwLock::new(metrics)),
+		Err(e) => {
+			if e.kind() == std::io::ErrorKind::NotFound {
+				Arc::new(RwLock::new(NodeMetrics::default()))
+			} else {
+				return Err(BuildError::ReadFailed);
+			}
+		},
+	};
+
 	// Initialize the on-chain wallet and chain access
 	let xprv = bitcoin::bip32::Xpriv::new_master(config.network, &seed_bytes).map_err(|e| {
 		log_error!(logger, "Failed to derive master secret: {}", e);
@@ -586,62 +608,54 @@ fn build_with_store_internal(
 			})?,
 	};
 
-	let (esplora_client, tx_sync, tx_broadcaster, fee_estimator) = match chain_data_source_config {
-		Some(ChainDataSourceConfig::Esplora(server_url)) => {
-			let mut client_builder = esplora_client::Builder::new(&server_url.clone());
-			client_builder = client_builder.timeout(DEFAULT_ESPLORA_CLIENT_TIMEOUT_SECS);
-			let esplora_client = client_builder.build_async().unwrap();
-			let tx_sync = Arc::new(EsploraSyncClient::from_client(
-				esplora_client.clone(),
-				Arc::clone(&logger),
-			));
-			let tx_broadcaster = Arc::new(TransactionBroadcaster::new(
-				tx_sync.client().clone(),
-				Arc::clone(&logger),
-			));
-			let fee_estimator = Arc::new(OnchainFeeEstimator::new(
-				tx_sync.client().clone(),
+	let tx_broadcaster = Arc::new(TransactionBroadcaster::new(Arc::clone(&logger)));
+	let fee_estimator = Arc::new(OnchainFeeEstimator::new());
+	let wallet = Arc::new(Wallet::new(
+		bdk_wallet,
+		wallet_persister,
+		Arc::clone(&tx_broadcaster),
+		Arc::clone(&fee_estimator),
+		Arc::clone(&logger),
+	));
+
+	let chain_source = match chain_data_source_config {
+		Some(ChainDataSourceConfig::Esplora { server_url, sync_config }) => {
+			let sync_config = sync_config.unwrap_or(EsploraSyncConfig::default());
+			Arc::new(ChainSource::new_esplora(
+				server_url.clone(),
+				sync_config,
+				Arc::clone(&wallet),
+				Arc::clone(&fee_estimator),
+				Arc::clone(&tx_broadcaster),
+				Arc::clone(&kv_store),
 				Arc::clone(&config),
 				Arc::clone(&logger),
-			));
-			(esplora_client, tx_sync, tx_broadcaster, fee_estimator)
+				Arc::clone(&node_metrics),
+			))
 		},
 		None => {
 			// Default to Esplora client.
 			let server_url = DEFAULT_ESPLORA_SERVER_URL.to_string();
-			let mut client_builder = esplora_client::Builder::new(&server_url);
-			client_builder = client_builder.timeout(DEFAULT_ESPLORA_CLIENT_TIMEOUT_SECS);
-			let esplora_client = client_builder.build_async().unwrap();
-			let tx_sync = Arc::new(EsploraSyncClient::from_client(
-				esplora_client.clone(),
-				Arc::clone(&logger),
-			));
-			let tx_broadcaster = Arc::new(TransactionBroadcaster::new(
-				tx_sync.client().clone(),
-				Arc::clone(&logger),
-			));
-			let fee_estimator = Arc::new(OnchainFeeEstimator::new(
-				tx_sync.client().clone(),
+			let sync_config = EsploraSyncConfig::default();
+			Arc::new(ChainSource::new_esplora(
+				server_url.clone(),
+				sync_config,
+				Arc::clone(&wallet),
+				Arc::clone(&fee_estimator),
+				Arc::clone(&tx_broadcaster),
+				Arc::clone(&kv_store),
 				Arc::clone(&config),
 				Arc::clone(&logger),
-			));
-			(esplora_client, tx_sync, tx_broadcaster, fee_estimator)
+				Arc::clone(&node_metrics),
+			))
 		},
 	};
 
 	let runtime = Arc::new(RwLock::new(None));
-	let wallet = Arc::new(Wallet::new(
-		bdk_wallet,
-		wallet_persister,
-		esplora_client,
-		Arc::clone(&tx_broadcaster),
-		Arc::clone(&fee_estimator),
-		Arc::clone(&logger),
-	));
 
 	// Initialize the ChainMonitor
 	let chain_monitor: Arc<ChainMonitor> = Arc::new(chainmonitor::ChainMonitor::new(
-		Some(Arc::clone(&tx_sync)),
+		Some(Arc::clone(&chain_source)),
 		Arc::clone(&tx_broadcaster),
 		Arc::clone(&logger),
 		Arc::clone(&fee_estimator),
@@ -824,23 +838,24 @@ fn build_with_store_internal(
 				Arc::new(GossipSource::new_p2p(Arc::clone(&network_graph), Arc::clone(&logger)));
 
 			// Reset the RGS sync timestamp in case we somehow switch gossip sources
-			io::utils::write_latest_rgs_sync_timestamp(
-				0,
-				Arc::clone(&kv_store),
-				Arc::clone(&logger),
-			)
-			.map_err(|e| {
-				log_error!(logger, "Failed writing to store: {}", e);
-				BuildError::WriteFailed
-			})?;
+			{
+				let mut locked_node_metrics = node_metrics.write().unwrap();
+				locked_node_metrics.latest_rgs_snapshot_timestamp = None;
+				write_node_metrics(
+					&*locked_node_metrics,
+					Arc::clone(&kv_store),
+					Arc::clone(&logger),
+				)
+				.map_err(|e| {
+					log_error!(logger, "Failed writing to store: {}", e);
+					BuildError::WriteFailed
+				})?;
+			}
 			p2p_source
 		},
 		GossipSourceConfig::RapidGossipSync(rgs_server) => {
-			let latest_sync_timestamp = io::utils::read_latest_rgs_sync_timestamp(
-				Arc::clone(&kv_store),
-				Arc::clone(&logger),
-			)
-			.unwrap_or(0);
+			let latest_sync_timestamp =
+				node_metrics.read().unwrap().latest_rgs_snapshot_timestamp.unwrap_or(0);
 			Arc::new(GossipSource::new_rgs(
 				rgs_server.clone(),
 				latest_sync_timestamp,
@@ -857,7 +872,7 @@ fn build_with_store_internal(
 			let liquidity_manager = Arc::new(LiquidityManager::new(
 				Arc::clone(&keys_manager),
 				Arc::clone(&channel_manager),
-				Some(Arc::clone(&tx_sync)),
+				Some(Arc::clone(&chain_source)),
 				None,
 				None,
 				liquidity_client_config,
@@ -925,7 +940,7 @@ fn build_with_store_internal(
 	let output_sweeper = match io::utils::read_output_sweeper(
 		Arc::clone(&tx_broadcaster),
 		Arc::clone(&fee_estimator),
-		Arc::clone(&tx_sync),
+		Arc::clone(&chain_source),
 		Arc::clone(&keys_manager),
 		Arc::clone(&kv_store),
 		Arc::clone(&logger),
@@ -937,7 +952,7 @@ fn build_with_store_internal(
 					channel_manager.current_best_block(),
 					Arc::clone(&tx_broadcaster),
 					Arc::clone(&fee_estimator),
-					Some(Arc::clone(&tx_sync)),
+					Some(Arc::clone(&chain_source)),
 					Arc::clone(&keys_manager),
 					Arc::clone(&keys_manager),
 					Arc::clone(&kv_store),
@@ -999,23 +1014,14 @@ fn build_with_store_internal(
 	let (stop_sender, _) = tokio::sync::watch::channel(());
 	let (event_handling_stopped_sender, _) = tokio::sync::watch::channel(());
 
-	let is_listening = Arc::new(AtomicBool::new(false));
-	let latest_wallet_sync_timestamp = Arc::new(RwLock::new(None));
-	let latest_onchain_wallet_sync_timestamp = Arc::new(RwLock::new(None));
-	let latest_fee_rate_cache_update_timestamp = Arc::new(RwLock::new(None));
-	let latest_rgs_snapshot_timestamp = Arc::new(RwLock::new(None));
-	let latest_node_announcement_broadcast_timestamp = Arc::new(RwLock::new(None));
-	let latest_channel_monitor_archival_height = Arc::new(RwLock::new(None));
-
 	Ok(Node {
 		runtime,
 		stop_sender,
 		event_handling_stopped_sender,
 		config,
 		wallet,
-		tx_sync,
+		chain_source,
 		tx_broadcaster,
-		fee_estimator,
 		event_queue,
 		channel_manager,
 		chain_monitor,
@@ -1034,12 +1040,7 @@ fn build_with_store_internal(
 		peer_store,
 		payment_store,
 		is_listening,
-		latest_wallet_sync_timestamp,
-		latest_onchain_wallet_sync_timestamp,
-		latest_fee_rate_cache_update_timestamp,
-		latest_rgs_snapshot_timestamp,
-		latest_node_announcement_broadcast_timestamp,
-		latest_channel_monitor_archival_height,
+		node_metrics,
 	})
 }