add timer

Author: joostjager

src/builder.rs

@@ -38,7 +38,8 @@ use lightning::ln::peer_handler::{IgnoringMessageHandler, MessageHandler};
 use lightning::routing::gossip::NodeAlias;
 use lightning::routing::router::DefaultRouter;
 use lightning::routing::scoring::{
-	CombinedScorer, ProbabilisticScorer, ProbabilisticScoringDecayParameters, ProbabilisticScoringFeeParameters
+	CombinedScorer, ProbabilisticScorer, ProbabilisticScoringDecayParameters,
+	ProbabilisticScoringFeeParameters,
 };
 use lightning::sign::EntropySource;
 
@@ -886,19 +887,15 @@ fn build_with_store_internal(
 		Err(e) => {
 			if e.kind() == std::io::ErrorKind::NotFound {
 				let params = ProbabilisticScoringDecayParameters::default();
-				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),
@@ -907,7 +904,7 @@ fn build_with_store_internal(
 		Arc::clone(&scorer),
 		scoring_fee_params,
 	));
-	
+
 	// Read ChannelMonitor state from store
 	let channel_monitors = match read_channel_monitors(
 		Arc::clone(&kv_store),

src/config.rs

@@ -50,6 +50,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_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);
 
@@ -71,6 +74,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_SCORES_SYNC_TIMEOUT_SECS: u64 = 5;
+
 // The length in bytes of our wallets' keys seed.
 pub(crate) const WALLET_KEYS_SEED_LEN: usize = 64;
 
@@ -162,6 +168,9 @@ pub struct Config {
 	/// **Note:** If unset, default parameters will be used, and you will be able to override the
 	/// parameters on a per-payment basis in the corresponding method calls.
 	pub sending_parameters: Option<SendingParameters>,
+
+	/// Optional URL to periodically fetch external scores from.
+	pub external_scores_url: Option<String>,
 }
 
 impl Default for Config {
@@ -177,6 +186,7 @@ impl Default for Config {
 			anchor_channels_config: Some(AnchorChannelsConfig::default()),
 			sending_parameters: None,
 			node_alias: None,
+			external_scores_url: None,
 		}
 	}
 }

src/lib.rs

@@ -99,6 +99,8 @@ mod wallet;
 pub use bip39;
 pub use bitcoin;
 pub use lightning;
+use lightning::routing::scoring::{ChannelLiquidities, CombinedScorer};
+use lightning::util::ser::Readable;
 pub use lightning_invoice;
 pub use lightning_types;
 pub use vss_client;
@@ -122,7 +124,8 @@ pub use builder::NodeBuilder as Builder;
 
 use chain::ChainSource;
 use config::{
-	default_user_config, may_announce_channel, ChannelConfig, Config, NODE_ANN_BCAST_INTERVAL,
+	default_user_config, may_announce_channel, ChannelConfig, Config,
+	EXTERNAL_SCORES_SYNC_INTERVAL, EXTERNAL_SCORES_SYNC_TIMEOUT_SECS, NODE_ANN_BCAST_INTERVAL,
 	PEER_RECONNECTION_INTERVAL, RGS_SYNC_INTERVAL,
 };
 use connection::ConnectionManager;
@@ -162,6 +165,7 @@ use bitcoin::secp256k1::PublicKey;
 use rand::Rng;
 
 use std::default::Default;
+use std::io::Cursor;
 use std::net::ToSocketAddrs;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex, RwLock};
@@ -306,6 +310,36 @@ impl Node {
 			});
 		}
 
+		if let Some(url) = &self.config.external_scores_url {
+			let external_scores_sync_logger = Arc::clone(&self.logger);
+			let external_scores_scorer = Arc::clone(&self.scorer);
+			let mut stop_sync = self.stop_sender.subscribe();
+			let url = url.clone();
+
+			runtime.spawn(async move {
+				let mut interval = tokio::time::interval(EXTERNAL_SCORES_SYNC_INTERVAL);
+				loop {
+					tokio::select! {
+						_ = stop_sync.changed() => {
+							log_trace!(
+								external_scores_sync_logger,
+								"Stopping background syncing external scores.",
+							);
+							return;
+						}
+						_ = interval.tick() => {
+							log_trace!(
+								external_scores_sync_logger,
+								"Background sync of external scores started.",
+							);
+
+							sync_external_scores(&external_scores_sync_logger, &external_scores_scorer, &url).await;
+						}
+					}
+				}
+			});
+		}
+
 		if let Some(listening_addresses) = &self.config.listening_addresses {
 			// Setup networking
 			let peer_manager_connection_handler = Arc::clone(&self.peer_manager);
@@ -1612,3 +1646,51 @@ pub(crate) fn total_anchor_channels_reserve_sats(
 			* anchor_channels_config.per_channel_reserve_sats
 	})
 }
+
+async fn sync_external_scores(
+	logger: &FilesystemLogger, scorer: &Mutex<CombinedScorer<Arc<Graph>, Arc<FilesystemLogger>>>,
+	url: &String,
+) -> () {
+	let response = tokio::time::timeout(
+		Duration::from_secs(EXTERNAL_SCORES_SYNC_TIMEOUT_SECS),
+		reqwest::get(url),
+	)
+	.await;
+
+	match response {
+		Ok(Ok(response)) => {
+			let body = response.bytes().await;
+			match body {
+				Err(e) => {
+					log_trace!(
+						logger,
+						"Failed to read external scores update from http source: {}",
+						e
+					);
+					return;
+				},
+				Ok(body) => {
+					let mut reader = Cursor::new(body);
+					match ChannelLiquidities::read(&mut reader) {
+						Ok(liquidities) => {
+							scorer.lock().unwrap().merge(liquidities);
+							log_trace!(logger, "External scores merged successfully",);
+						},
+						Err(e) => {
+							log_trace!(logger, "Failed to parse external scores update: {}", e);
+							return;
+						},
+					}
+				},
+			}
+		},
+		Err(e) => {
+			log_trace!(logger, "Retrieving external scores timed out: {}", e);
+			return;
+		},
+		Ok(Err(e)) => {
+			log_trace!(logger, "Failed to retrieve external scores update: {}", e);
+			return;
+		},
+	}
+}

src/types.rs

@@ -22,7 +22,7 @@ use lightning::ln::types::ChannelId;
 use lightning::routing::gossip;
 use lightning::routing::router::DefaultRouter;
 use lightning::routing::scoring::CombinedScorer;
-use lightning::routing::scoring::{ProbabilisticScorer, ProbabilisticScoringFeeParameters};
+use lightning::routing::scoring::ProbabilisticScoringFeeParameters;
 use lightning::sign::InMemorySigner;
 use lightning::util::persist::KVStore;
 use lightning::util::ser::{Readable, Writeable, Writer};