add functionality to periodically update routing scores from an external http source

Author: joostjager

bindings/ldk_node.udl

@@ -58,6 +58,7 @@ interface Builder {
 	void set_chain_source_bitcoind_rpc(string rpc_host, u16 rpc_port, string rpc_user, string rpc_password);
 	void set_gossip_source_p2p();
 	void set_gossip_source_rgs(string rgs_server_url);
+	void set_pathfinding_scores_source(string url);
 	void set_liquidity_source_lsps2(SocketAddress address, PublicKey node_id, string? token);
 	void set_storage_dir_path(string storage_dir_path);
 	void set_filesystem_logger(string? log_file_path, LogLevel? log_level);
@@ -273,6 +274,7 @@ dictionary NodeStatus {
 	u64? latest_onchain_wallet_sync_timestamp;
 	u64? latest_fee_rate_cache_update_timestamp;
 	u64? latest_rgs_snapshot_timestamp;
+	u64? latest_pathfinding_scores_sync_timestamp;
 	u64? latest_node_announcement_broadcast_timestamp;
 	u32? latest_channel_monitor_archival_height;
 };

src/builder.rs

@@ -30,8 +30,8 @@ use crate::types::{
 };
 use crate::wallet::persist::KVStoreWalletPersister;
 use crate::wallet::Wallet;
-use crate::Node;
 use crate::{io, NodeMetrics};
+use crate::{Node, PathfindingScoresSyncConfig};
 
 use lightning::chain::{chainmonitor, BestBlock, Watch};
 use lightning::io::Cursor;
@@ -41,7 +41,8 @@ use lightning::ln::peer_handler::{IgnoringMessageHandler, MessageHandler};
 use lightning::routing::gossip::NodeAlias;
 use lightning::routing::router::DefaultRouter;
 use lightning::routing::scoring::{
-	ProbabilisticScorer, ProbabilisticScoringDecayParameters, ProbabilisticScoringFeeParameters,
+	CombinedScorer, ProbabilisticScorer, ProbabilisticScoringDecayParameters,
+	ProbabilisticScoringFeeParameters,
 };
 use lightning::sign::EntropySource;
 
@@ -229,6 +230,7 @@ pub struct NodeBuilder {
 	gossip_source_config: Option<GossipSourceConfig>,
 	liquidity_source_config: Option<LiquiditySourceConfig>,
 	log_writer_config: Option<LogWriterConfig>,
+	pathfinding_scores_sync_config: Option<PathfindingScoresSyncConfig>,
 }
 
 impl NodeBuilder {
@@ -245,13 +247,15 @@ impl NodeBuilder {
 		let gossip_source_config = None;
 		let liquidity_source_config = None;
 		let log_writer_config = None;
+		let scoring_source_config = None;
 		Self {
 			config,
 			entropy_source_config,
 			chain_data_source_config,
 			gossip_source_config,
 			liquidity_source_config,
 			log_writer_config,
+			pathfinding_scores_sync_config: scoring_source_config,
 		}
 	}
 
@@ -322,6 +326,14 @@ impl NodeBuilder {
 		self
 	}
 
+	/// Configures the [`Node`] instance to source its external scores from the given URL.
+	///
+	/// The external scores are merged into the local scoring system to improve routing.
+	pub fn set_pathfinding_scores_source(&mut self, url: String) -> &mut Self {
+		self.pathfinding_scores_sync_config = Some(PathfindingScoresSyncConfig { url });
+		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.
@@ -540,6 +552,7 @@ impl NodeBuilder {
 			config,
 			self.chain_data_source_config.as_ref(),
 			self.gossip_source_config.as_ref(),
+			self.pathfinding_scores_sync_config.as_ref(),
 			self.liquidity_source_config.as_ref(),
 			seed_bytes,
 			logger,
@@ -562,6 +575,7 @@ impl NodeBuilder {
 			config,
 			self.chain_data_source_config.as_ref(),
 			self.gossip_source_config.as_ref(),
+			self.pathfinding_scores_sync_config.as_ref(),
 			self.liquidity_source_config.as_ref(),
 			seed_bytes,
 			logger,
@@ -654,6 +668,12 @@ impl ArcedNodeBuilder {
 		self.inner.write().unwrap().set_gossip_source_rgs(rgs_server_url);
 	}
 
+	/// Configures the [`Node`] instance to source its external scores from the given URL. These scores are used to
+	/// augment the internal pathfinding scoring system to improve routing.
+	pub fn set_pathfinding_scores_source(&self, url: String) {
+		self.inner.write().unwrap().set_pathfinding_scores_source(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.
@@ -806,6 +826,7 @@ impl ArcedNodeBuilder {
 fn build_with_store_internal(
 	config: Arc<Config>, chain_data_source_config: Option<&ChainDataSourceConfig>,
 	gossip_source_config: Option<&GossipSourceConfig>,
+	pathfinding_scores_sync_config: Option<&PathfindingScoresSyncConfig>,
 	liquidity_source_config: Option<&LiquiditySourceConfig>, seed_bytes: [u8; 64],
 	logger: Arc<Logger>, kv_store: Arc<DynStore>,
 ) -> Result<Node, BuildError> {
@@ -950,26 +971,24 @@ fn build_with_store_internal(
 			},
 		};
 
-	let scorer = match io::utils::read_scorer(
+	let local_scorer = match io::utils::read_scorer(
 		Arc::clone(&kv_store),
 		Arc::clone(&network_graph),
 		Arc::clone(&logger),
 	) {
-		Ok(scorer) => Arc::new(Mutex::new(scorer)),
+		Ok(scorer) => scorer,
 		Err(e) => {
 			if e.kind() == std::io::ErrorKind::NotFound {
 				let params = ProbabilisticScoringDecayParameters::default();
-				Arc::new(Mutex::new(ProbabilisticScorer::new(
-					params,
-					Arc::clone(&network_graph),
-					Arc::clone(&logger),
-				)))
+				ProbabilisticScorer::new(params, Arc::clone(&network_graph), Arc::clone(&logger))
 			} else {
 				return Err(BuildError::ReadFailed);
 			}
 		},
 	};
 
+	let scorer = Arc::new(Mutex::new(CombinedScorer::new(local_scorer)));
+
 	let scoring_fee_params = ProbabilisticScoringFeeParameters::default();
 	let router = Arc::new(DefaultRouter::new(
 		Arc::clone(&network_graph),
@@ -1303,6 +1322,7 @@ fn build_with_store_internal(
 		keys_manager,
 		network_graph,
 		gossip_source,
+		pathfinding_scores_sync_config: pathfinding_scores_sync_config.cloned(),
 		liquidity_source,
 		kv_store,
 		logger,

src/config.rs

@@ -57,6 +57,9 @@ pub(crate) const PEER_RECONNECTION_INTERVAL: Duration = Duration::from_secs(10);
 // The time in-between RGS sync attempts.
 pub(crate) const RGS_SYNC_INTERVAL: Duration = Duration::from_secs(60 * 60);
 
+// The time in-between external scores sync attempts.
+pub(crate) const EXTERNAL_PATHFINDING_SCORES_SYNC_INTERVAL: Duration = Duration::from_secs(60 * 60);
+
 // The time in-between node announcement broadcast attempts.
 pub(crate) const NODE_ANN_BCAST_INTERVAL: Duration = Duration::from_secs(60 * 60);
 
@@ -78,6 +81,9 @@ pub(crate) const TX_BROADCAST_TIMEOUT_SECS: u64 = 5;
 // The timeout after which we abort a RGS sync operation.
 pub(crate) const RGS_SYNC_TIMEOUT_SECS: u64 = 5;
 
+// The timeout after which we abort a external scores sync operation.
+pub(crate) const EXTERNAL_PATHFINDING_SCORES_SYNC_TIMEOUT_SECS: u64 = 5;
+
 // The length in bytes of our wallets' keys seed.
 pub(crate) const WALLET_KEYS_SEED_LEN: usize = 64;
 

src/lib.rs

@@ -89,6 +89,7 @@ pub mod logger;
 mod message_handler;
 pub mod payment;
 mod peer_store;
+mod scoring;
 mod sweep;
 mod tx_broadcaster;
 mod types;
@@ -122,8 +123,9 @@ pub use builder::NodeBuilder as Builder;
 
 use chain::ChainSource;
 use config::{
-	default_user_config, may_announce_channel, ChannelConfig, Config, NODE_ANN_BCAST_INTERVAL,
-	PEER_RECONNECTION_INTERVAL, RGS_SYNC_INTERVAL,
+	default_user_config, may_announce_channel, ChannelConfig, Config,
+	EXTERNAL_PATHFINDING_SCORES_SYNC_INTERVAL, EXTERNAL_PATHFINDING_SCORES_SYNC_TIMEOUT_SECS,
+	NODE_ANN_BCAST_INTERVAL, PEER_RECONNECTION_INTERVAL, RGS_SYNC_INTERVAL,
 };
 use connection::ConnectionManager;
 use event::{EventHandler, EventQueue};
@@ -137,6 +139,7 @@ use payment::{
 	UnifiedQrPayment,
 };
 use peer_store::{PeerInfo, PeerStore};
+use scoring::setup_background_pathfinding_scores_sync;
 use types::{
 	Broadcaster, BumpTransactionEventHandler, ChainMonitor, ChannelManager, DynStore, Graph,
 	KeysManager, OnionMessenger, PeerManager, Router, Scorer, Sweeper, Wallet,
@@ -169,6 +172,11 @@ use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 #[cfg(feature = "uniffi")]
 uniffi::include_scaffolding!("ldk_node");
 
+#[derive(Debug, Clone)]
+struct PathfindingScoresSyncConfig {
+	url: String,
+}
+
 /// The main interface object of LDK Node, wrapping the necessary LDK and BDK functionalities.
 ///
 /// Needs to be initialized and instantiated through [`Builder::build`].
@@ -190,6 +198,7 @@ pub struct Node {
 	keys_manager: Arc<KeysManager>,
 	network_graph: Arc<Graph>,
 	gossip_source: Arc<GossipSource>,
+	pathfinding_scores_sync_config: Option<PathfindingScoresSyncConfig>,
 	liquidity_source: Option<Arc<LiquiditySource<Arc<Logger>>>>,
 	kv_store: Arc<DynStore>,
 	logger: Arc<Logger>,
@@ -304,6 +313,18 @@ impl Node {
 			});
 		}
 
+		if let Some(pathfinding_scores_sync_config) = self.pathfinding_scores_sync_config.as_ref() {
+			setup_background_pathfinding_scores_sync(
+				pathfinding_scores_sync_config.url.clone(),
+				self.scorer.clone(),
+				self.node_metrics.clone(),
+				self.kv_store.clone(),
+				self.logger.clone(),
+				&runtime,
+				&self.stop_sender,
+			);
+		}
+
 		if let Some(listening_addresses) = &self.config.listening_addresses {
 			// Setup networking
 			let peer_manager_connection_handler = Arc::clone(&self.peer_manager);
@@ -725,6 +746,8 @@ impl Node {
 			locked_node_metrics.latest_fee_rate_cache_update_timestamp;
 		let latest_rgs_snapshot_timestamp =
 			locked_node_metrics.latest_rgs_snapshot_timestamp.map(|val| val as u64);
+		let latest_pathfinding_scores_sync_timestamp =
+			locked_node_metrics.latest_pathfinding_scores_sync_timestamp;
 		let latest_node_announcement_broadcast_timestamp =
 			locked_node_metrics.latest_node_announcement_broadcast_timestamp;
 		let latest_channel_monitor_archival_height =
@@ -738,6 +761,7 @@ impl Node {
 			latest_onchain_wallet_sync_timestamp,
 			latest_fee_rate_cache_update_timestamp,
 			latest_rgs_snapshot_timestamp,
+			latest_pathfinding_scores_sync_timestamp,
 			latest_node_announcement_broadcast_timestamp,
 			latest_channel_monitor_archival_height,
 		}
@@ -1547,6 +1571,8 @@ pub struct NodeStatus {
 	///
 	/// Will be `None` if RGS isn't configured or the snapshot hasn't been updated yet.
 	pub latest_rgs_snapshot_timestamp: Option<u64>,
+	/// The timestamp, in seconds since start of the UNIX epoch, when we last successfully merged external scores.
+	pub latest_pathfinding_scores_sync_timestamp: Option<u64>,
 	/// The timestamp, in seconds since start of the UNIX epoch, when we last broadcasted a node
 	/// announcement.
 	///
@@ -1565,6 +1591,7 @@ pub(crate) struct NodeMetrics {
 	latest_onchain_wallet_sync_timestamp: Option<u64>,
 	latest_fee_rate_cache_update_timestamp: Option<u64>,
 	latest_rgs_snapshot_timestamp: Option<u32>,
+	latest_pathfinding_scores_sync_timestamp: Option<u64>,
 	latest_node_announcement_broadcast_timestamp: Option<u64>,
 	latest_channel_monitor_archival_height: Option<u32>,
 }
@@ -1576,6 +1603,7 @@ impl Default for NodeMetrics {
 			latest_onchain_wallet_sync_timestamp: None,
 			latest_fee_rate_cache_update_timestamp: None,
 			latest_rgs_snapshot_timestamp: None,
+			latest_pathfinding_scores_sync_timestamp: None,
 			latest_node_announcement_broadcast_timestamp: None,
 			latest_channel_monitor_archival_height: None,
 		}
@@ -1584,6 +1612,7 @@ impl Default for NodeMetrics {
 
 impl_writeable_tlv_based!(NodeMetrics, {
 	(0, latest_lightning_wallet_sync_timestamp, option),
+	(1, latest_pathfinding_scores_sync_timestamp, option),
 	(2, latest_onchain_wallet_sync_timestamp, option),
 	(4, latest_fee_rate_cache_update_timestamp, option),
 	(6, latest_rgs_snapshot_timestamp, option),

src/scoring.rs

@@ -0,0 +1,105 @@
+use std::{
+	io::Cursor,
+	sync::{Arc, Mutex, RwLock},
+	time::{Duration, SystemTime},
+};
+
+use crate::{logger::LdkLogger, NodeMetrics};
+use crate::{
+	write_node_metrics, DynStore, Graph, Logger, EXTERNAL_PATHFINDING_SCORES_SYNC_INTERVAL,
+	EXTERNAL_PATHFINDING_SCORES_SYNC_TIMEOUT_SECS,
+};
+use lightning::{
+	log_error, log_info, log_trace,
+	routing::scoring::{ChannelLiquidities, CombinedScorer},
+	util::ser::Readable,
+};
+
+/// Start a background task that periodically downloads scores via an external url and merges them into the local
+/// pathfinding scores.
+pub fn setup_background_pathfinding_scores_sync(
+	url: String, scorer: Arc<Mutex<CombinedScorer<Arc<Graph>, Arc<Logger>>>>,
+	node_metrics: Arc<RwLock<NodeMetrics>>, kv_store: Arc<DynStore>, logger: Arc<Logger>,
+	runtime: &tokio::runtime::Runtime, stop_sender: &tokio::sync::watch::Sender<()>,
+) {
+	log_info!(logger, "External scores background syncing from {} enabled", url);
+
+	let logger = Arc::clone(&logger);
+	let mut stop_sync = stop_sender.subscribe();
+
+	runtime.spawn(async move {
+		let mut interval = tokio::time::interval(EXTERNAL_PATHFINDING_SCORES_SYNC_INTERVAL);
+		loop {
+			tokio::select! {
+				_ = stop_sync.changed() => {
+					log_trace!(
+						logger,
+						"Stopping background syncing external scores.",
+					);
+					return;
+				}
+				_ = interval.tick() => {
+					log_trace!(
+						logger,
+						"Background sync of external scores started.",
+					);
+
+					sync_external_scores(logger.as_ref(), scorer.as_ref(), node_metrics.as_ref(), Arc::clone(&kv_store), &url).await;
+				}
+			}
+		}
+	});
+}
+
+async fn sync_external_scores(
+	logger: &Logger, scorer: &Mutex<CombinedScorer<Arc<Graph>, Arc<Logger>>>,
+	node_metrics: &RwLock<NodeMetrics>, kv_store: Arc<DynStore>, url: &String,
+) -> () {
+	let response = tokio::time::timeout(
+		Duration::from_secs(EXTERNAL_PATHFINDING_SCORES_SYNC_TIMEOUT_SECS),
+		reqwest::get(url),
+	)
+	.await;
+
+	match response {
+		Ok(Ok(response)) => {
+			let body = response.bytes().await;
+			match body {
+				Err(e) => {
+					log_error!(logger, "Failed to read external scores update: {}", e);
+					return;
+				},
+				Ok(body) => {
+					let mut reader = Cursor::new(body);
+					match ChannelLiquidities::read(&mut reader) {
+						Ok(liquidities) => {
+							let duration_since_epoch =
+								SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap();
+							scorer.lock().unwrap().merge(liquidities, duration_since_epoch);
+							let mut locked_node_metrics = node_metrics.write().unwrap();
+							locked_node_metrics.latest_pathfinding_scores_sync_timestamp =
+								Some(duration_since_epoch.as_secs());
+							write_node_metrics(&*locked_node_metrics, kv_store, logger)
+								.unwrap_or_else(|e| {
+									log_error!(logger, "Persisting node metrics failed: {}", e);
+								});
+							log_trace!(logger, "External scores merged successfully");
+						},
+						Err(e) => {
+							log_error!(logger, "Failed to parse external scores update: {}", e);
+							return;
+						},
+					}
+				},
+			}
+		},
+		Err(e) => {
+			log_error!(logger, "Retrieving external scores timed out: {}", e);
+			return;
+		},
+		Ok(Err(e)) => {
+			log_error!(logger, "Failed to retrieve external scores update: {}", e);
+			return;
+		},
+	}
+}