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

Author: joostjager

bindings/ldk_node.udl

@@ -81,6 +81,7 @@ interface Builder {
 	void set_chain_source_bitcoind_rest(string rest_host, u16 rest_port, 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_lsps1(PublicKey node_id, SocketAddress address, string? token);
 	void set_liquidity_source_lsps2(PublicKey node_id, SocketAddress address, string? token);
 	void set_storage_dir_path(string storage_dir_path);
@@ -330,6 +331,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

@@ -27,7 +27,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, NodeSigner};
 use lightning::util::persist::{
@@ -110,6 +111,11 @@ enum GossipSourceConfig {
 	RapidGossipSync(String),
 }
 
+#[derive(Debug, Clone)]
+struct PathfindingScoresSyncConfig {
+	url: String,
+}
+
 #[derive(Debug, Clone, Default)]
 struct LiquiditySourceConfig {
 	// Act as an LSPS1 client connecting to the given service.
@@ -243,6 +249,7 @@ pub struct NodeBuilder {
 	log_writer_config: Option<LogWriterConfig>,
 	async_payments_role: Option<AsyncPaymentsRole>,
 	runtime_handle: Option<tokio::runtime::Handle>,
+	pathfinding_scores_sync_config: Option<PathfindingScoresSyncConfig>,
 }
 
 impl NodeBuilder {
@@ -260,6 +267,7 @@ impl NodeBuilder {
 		let liquidity_source_config = None;
 		let log_writer_config = None;
 		let runtime_handle = None;
+		let pathfinding_scores_sync_config = None;
 		Self {
 			config,
 			entropy_source_config,
@@ -269,6 +277,7 @@ impl NodeBuilder {
 			log_writer_config,
 			runtime_handle,
 			async_payments_role: None,
+			pathfinding_scores_sync_config,
 		}
 	}
 
@@ -411,6 +420,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 inbound liquidity from the given
 	/// [bLIP-51 / LSPS1] service.
 	///
@@ -718,6 +735,7 @@ impl NodeBuilder {
 			self.chain_data_source_config.as_ref(),
 			self.gossip_source_config.as_ref(),
 			self.liquidity_source_config.as_ref(),
+			self.pathfinding_scores_sync_config.as_ref(),
 			self.async_payments_role,
 			seed_bytes,
 			runtime,
@@ -751,6 +769,7 @@ impl NodeBuilder {
 			self.chain_data_source_config.as_ref(),
 			self.gossip_source_config.as_ref(),
 			self.liquidity_source_config.as_ref(),
+			self.pathfinding_scores_sync_config.as_ref(),
 			self.async_payments_role,
 			seed_bytes,
 			runtime,
@@ -910,6 +929,13 @@ 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.
+	///
+	/// The external scores are merged into the local 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 inbound liquidity from the given
 	/// [bLIP-51 / LSPS1] service.
 	///
@@ -1110,6 +1136,7 @@ fn build_with_store_internal(
 	config: Arc<Config>, chain_data_source_config: Option<&ChainDataSourceConfig>,
 	gossip_source_config: Option<&GossipSourceConfig>,
 	liquidity_source_config: Option<&LiquiditySourceConfig>,
+	pathfinding_scores_sync_config: Option<&PathfindingScoresSyncConfig>,
 	async_payments_role: Option<AsyncPaymentsRole>, seed_bytes: [u8; 64], runtime: Arc<Runtime>,
 	logger: Arc<Logger>, kv_store: Arc<DynStore>,
 ) -> Result<Node, BuildError> {
@@ -1365,26 +1392,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),
@@ -1716,6 +1741,8 @@ fn build_with_store_internal(
 	let (background_processor_stop_sender, _) = tokio::sync::watch::channel(());
 	let is_running = Arc::new(RwLock::new(false));
 
+	let pathfinding_scores_sync_url = pathfinding_scores_sync_config.map(|c| c.url.clone());
+
 	Ok(Node {
 		runtime,
 		stop_sender,
@@ -1734,6 +1761,7 @@ fn build_with_store_internal(
 		keys_manager,
 		network_graph,
 		gossip_source,
+		pathfinding_scores_sync_url,
 		liquidity_source,
 		kv_store,
 		logger,

src/config.rs

@@ -63,6 +63,9 @@ pub(crate) const PEER_RECONNECTION_INTERVAL: Duration = Duration::from_secs(60);
 // 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);
 
@@ -93,6 +96,9 @@ pub(crate) const RGS_SYNC_TIMEOUT_SECS: u64 = 5;
 /// The length in bytes of our wallets' keys seed.
 pub const WALLET_KEYS_SEED_LEN: usize = 64;
 
+// The timeout after which we abort a external scores sync operation.
+pub(crate) const EXTERNAL_PATHFINDING_SCORES_SYNC_TIMEOUT_SECS: u64 = 5;
+
 #[derive(Debug, Clone)]
 /// Represents the configuration of an [`Node`] instance.
 ///

src/lib.rs

@@ -95,6 +95,7 @@ mod message_handler;
 pub mod payment;
 mod peer_store;
 mod runtime;
+mod scoring;
 mod tx_broadcaster;
 mod types;
 mod wallet;
@@ -104,6 +105,7 @@ use std::net::ToSocketAddrs;
 use std::sync::{Arc, Mutex, RwLock};
 use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 
+use crate::scoring::setup_background_pathfinding_scores_sync;
 pub use balance::{BalanceDetails, LightningBalance, PendingSweepBalance};
 use bitcoin::secp256k1::PublicKey;
 #[cfg(feature = "uniffi")]
@@ -154,6 +156,7 @@ use types::{
 pub use types::{
 	ChannelDetails, CustomTlvRecord, DynStore, PeerDetails, SyncAndAsyncKVStore, UserChannelId,
 };
+
 pub use {
 	bip39, bitcoin, lightning, lightning_invoice, lightning_liquidity, lightning_types, tokio,
 	vss_client,
@@ -183,6 +186,7 @@ pub struct Node {
 	keys_manager: Arc<KeysManager>,
 	network_graph: Arc<Graph>,
 	gossip_source: Arc<GossipSource>,
+	pathfinding_scores_sync_url: Option<String>,
 	liquidity_source: Option<Arc<LiquiditySource<Arc<Logger>>>>,
 	kv_store: Arc<DynStore>,
 	logger: Arc<Logger>,
@@ -290,6 +294,18 @@ impl Node {
 			});
 		}
 
+		if let Some(pathfinding_scores_sync_url) = self.pathfinding_scores_sync_url.as_ref() {
+			setup_background_pathfinding_scores_sync(
+				pathfinding_scores_sync_url.clone(),
+				Arc::clone(&self.scorer),
+				Arc::clone(&self.node_metrics),
+				Arc::clone(&self.kv_store),
+				Arc::clone(&self.logger),
+				Arc::clone(&self.runtime),
+				self.stop_sender.subscribe(),
+			);
+		}
+
 		if let Some(listening_addresses) = &self.config.listening_addresses {
 			// Setup networking
 			let peer_manager_connection_handler = Arc::clone(&self.peer_manager);
@@ -691,6 +707,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 =
@@ -703,6 +721,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,
 		}
@@ -1530,6 +1549,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.
 	///
@@ -1548,6 +1569,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>,
 }
@@ -1559,6 +1581,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,
 		}
@@ -1567,6 +1590,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,100 @@
+use std::{
+	io::Cursor,
+	sync::{Arc, Mutex, RwLock},
+	time::{Duration, SystemTime},
+};
+
+use crate::{
+	config::{
+		EXTERNAL_PATHFINDING_SCORES_SYNC_INTERVAL, EXTERNAL_PATHFINDING_SCORES_SYNC_TIMEOUT_SECS,
+	},
+	logger::LdkLogger,
+	runtime::Runtime,
+	NodeMetrics, Scorer,
+};
+use crate::{write_node_metrics, DynStore, Logger};
+use lightning::{
+	log_error, log_info, log_trace, routing::scoring::ChannelLiquidities, 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<crate::types::Scorer>>, node_metrics: Arc<RwLock<NodeMetrics>>,
+	kv_store: Arc<DynStore>, logger: Arc<Logger>, runtime: Arc<Runtime>,
+	mut stop_receiver: tokio::sync::watch::Receiver<()>,
+) {
+	log_info!(logger, "External scores background syncing enabled from {}", url);
+
+	let logger = Arc::clone(&logger);
+
+	runtime.spawn_background_processor_task(async move {
+		let mut interval = tokio::time::interval(EXTERNAL_PATHFINDING_SCORES_SYNC_INTERVAL);
+		loop {
+			tokio::select! {
+				_ = stop_receiver.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<Scorer>, 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;
+
+	if let Err(e) = response {
+		log_error!(logger, "Retrieving external scores timed out: {}", e);
+		return;
+	}
+	let response = response.unwrap();
+
+	if let Err(e) = response {
+		log_error!(logger, "Failed to retrieve external scores update: {}", e);
+		return;
+	}
+
+	let body = response.unwrap().bytes().await;
+	if let Err(e) = body {
+		log_error!(logger, "Failed to read external scores update: {}", e);
+		return;
+	}
+
+	let mut reader = Cursor::new(body.unwrap());
+	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);
+		},
+	}
+}