Merge pull request #449 from joostjager/import-scores

Periodical external pathfinding scores merge

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

@@ -24,10 +24,12 @@ use lightning::io::Cursor;
 use lightning::ln::channelmanager::{self, ChainParameters, ChannelManagerReadArgs};
 use lightning::ln::msgs::{RoutingMessageHandler, SocketAddress};
 use lightning::ln::peer_handler::{IgnoringMessageHandler, MessageHandler};
+use lightning::log_trace;
 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::{
@@ -50,7 +52,9 @@ use crate::event::EventQueue;
 use crate::fee_estimator::OnchainFeeEstimator;
 use crate::gossip::GossipSource;
 use crate::io::sqlite_store::SqliteStore;
-use crate::io::utils::{read_node_metrics, write_node_metrics};
+use crate::io::utils::{
+	read_external_pathfinding_scores_from_cache, read_node_metrics, write_node_metrics,
+};
 use crate::io::vss_store::VssStore;
 use crate::io::{
 	self, PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
@@ -110,6 +114,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 +252,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 +270,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 +280,7 @@ impl NodeBuilder {
 			log_writer_config,
 			runtime_handle,
 			async_payments_role: None,
+			pathfinding_scores_sync_config,
 		}
 	}
 
@@ -411,6 +423,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 +738,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 +772,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 +932,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 +1139,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 +1395,38 @@ 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)));
+
+	// Restore external pathfinding scores from cache if possible.
+	match read_external_pathfinding_scores_from_cache(Arc::clone(&kv_store), Arc::clone(&logger)) {
+		Ok(external_scores) => {
+			scorer.lock().unwrap().merge(external_scores, cur_time);
+			log_trace!(logger, "External scores from cache merged successfully");
+		},
+		Err(e) => {
+			if e.kind() != std::io::ErrorKind::NotFound {
+				log_error!(logger, "Error while reading external scores from cache: {}", e);
+				return Err(BuildError::ReadFailed);
+			}
+		},
+	}
+
 	let scoring_fee_params = ProbabilisticScoringFeeParameters::default();
 	let router = Arc::new(DefaultRouter::new(
 		Arc::clone(&network_graph),
@@ -1716,6 +1758,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 +1778,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/io/utils.rs

@@ -22,9 +22,11 @@ use bitcoin::Network;
 use lightning::io::Cursor;
 use lightning::ln::msgs::DecodeError;
 use lightning::routing::gossip::NetworkGraph;
-use lightning::routing::scoring::{ProbabilisticScorer, ProbabilisticScoringDecayParameters};
+use lightning::routing::scoring::{
+	ChannelLiquidities, ProbabilisticScorer, ProbabilisticScoringDecayParameters,
+};
 use lightning::util::persist::{
-	KVStoreSync, KVSTORE_NAMESPACE_KEY_ALPHABET, KVSTORE_NAMESPACE_KEY_MAX_LEN,
+	KVStore, KVStoreSync, KVSTORE_NAMESPACE_KEY_ALPHABET, KVSTORE_NAMESPACE_KEY_MAX_LEN,
 	NETWORK_GRAPH_PERSISTENCE_KEY, NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE,
 	NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, OUTPUT_SWEEPER_PERSISTENCE_KEY,
 	OUTPUT_SWEEPER_PERSISTENCE_PRIMARY_NAMESPACE, OUTPUT_SWEEPER_PERSISTENCE_SECONDARY_NAMESPACE,
@@ -48,6 +50,8 @@ use crate::types::{Broadcaster, DynStore, KeysManager, Sweeper};
 use crate::wallet::ser::{ChangeSetDeserWrapper, ChangeSetSerWrapper};
 use crate::{Error, EventQueue, NodeMetrics, PaymentDetails};
 
+pub const EXTERNAL_PATHFINDING_SCORES_CACHE_KEY: &str = "external_pathfinding_scores_cache";
+
 /// Generates a random [BIP 39] mnemonic.
 ///
 /// The result may be used to initialize the [`Node`] entropy, i.e., can be given to
@@ -164,6 +168,53 @@ where
 	})
 }
 
+/// Read previously persisted external pathfinding scores from the cache.
+pub(crate) fn read_external_pathfinding_scores_from_cache<L: Deref>(
+	kv_store: Arc<DynStore>, logger: L,
+) -> Result<ChannelLiquidities, std::io::Error>
+where
+	L::Target: LdkLogger,
+{
+	let mut reader = Cursor::new(KVStoreSync::read(
+		&*kv_store,
+		SCORER_PERSISTENCE_PRIMARY_NAMESPACE,
+		SCORER_PERSISTENCE_SECONDARY_NAMESPACE,
+		EXTERNAL_PATHFINDING_SCORES_CACHE_KEY,
+	)?);
+	ChannelLiquidities::read(&mut reader).map_err(|e| {
+		log_error!(logger, "Failed to deserialize scorer: {}", e);
+		std::io::Error::new(std::io::ErrorKind::InvalidData, "Failed to deserialize Scorer")
+	})
+}
+
+/// Persist external pathfinding scores to the cache.
+pub(crate) async fn write_external_pathfinding_scores_to_cache<L: Deref>(
+	kv_store: Arc<DynStore>, data: &ChannelLiquidities, logger: L,
+) -> Result<(), Error>
+where
+	L::Target: LdkLogger,
+{
+	KVStore::write(
+		&*kv_store,
+		SCORER_PERSISTENCE_PRIMARY_NAMESPACE,
+		SCORER_PERSISTENCE_SECONDARY_NAMESPACE,
+		EXTERNAL_PATHFINDING_SCORES_CACHE_KEY,
+		data.encode(),
+	)
+	.await
+	.map_err(|e| {
+		log_error!(
+			logger,
+			"Writing data to key {}/{}/{} failed due to: {}",
+			NODE_METRICS_PRIMARY_NAMESPACE,
+			NODE_METRICS_SECONDARY_NAMESPACE,
+			EXTERNAL_PATHFINDING_SCORES_CACHE_KEY,
+			e
+		);
+		Error::PersistenceFailed
+	})
+}
+
 /// Read previously persisted events from the store.
 pub(crate) fn read_event_queue<L: Deref + Clone>(
 	kv_store: Arc<DynStore>, logger: L,

src/lib.rs

@@ -97,6 +97,7 @@ mod message_handler;
 pub mod payment;
 mod peer_store;
 mod runtime;
+mod scoring;
 mod tx_broadcaster;
 mod types;
 mod wallet;
@@ -106,6 +107,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")]
@@ -156,6 +158,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,
@@ -185,6 +188,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>,
@@ -261,7 +265,6 @@ impl Node {
 							return;
 						}
 						_ = interval.tick() => {
-							let gossip_sync_logger = Arc::clone(&gossip_sync_logger);
 							let now = Instant::now();
 							match gossip_source.update_rgs_snapshot().await {
 								Ok(updated_timestamp) => {
@@ -293,6 +296,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);
@@ -694,6 +709,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 =
@@ -706,6 +723,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,
 		}
@@ -1533,6 +1551,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.
 	///
@@ -1551,6 +1571,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>,
 }
@@ -1562,6 +1583,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,
 		}
@@ -1570,6 +1592,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),