f Update GossipSource task to use select

Author: tnull

src/io/mod.rs

@@ -27,8 +27,8 @@ pub(crate) const PEER_INFO_PERSISTENCE_KEY: &str = "peers";
 pub(crate) const PAYMENT_INFO_PERSISTENCE_NAMESPACE: &str = "payments";
 
 /// RapidGossipSync's `latest_sync_timestamp` will be persisted under this key.
-pub(crate) const RGS_LATEST_SYNC_TIMESTAMP_NAMESPACE: &str = "";
-pub(crate) const RGS_LATEST_SYNC_TIMESTAMP_KEY: &str = "rgs_latest_sync_timestamp";
+pub(crate) const LATEST_RGS_SYNC_TIMESTAMP_NAMESPACE: &str = "";
+pub(crate) const LATEST_RGS_SYNC_TIMESTAMP_KEY: &str = "latest_rgs_sync_timestamp";
 
 /// Provides an interface that allows to store and retrieve persisted values that are associated
 /// with given keys.

src/io/utils.rs

@@ -174,12 +174,12 @@ where
 	Ok(res)
 }
 
-pub(crate) fn read_rgs_latest_sync_timestamp<K: Deref>(kv_store: K) -> Result<u32, std::io::Error>
+pub(crate) fn read_latest_rgs_sync_timestamp<K: Deref>(kv_store: K) -> Result<u32, std::io::Error>
 where
 	K::Target: KVStore,
 {
 	let mut reader =
-		kv_store.read(RGS_LATEST_SYNC_TIMESTAMP_NAMESPACE, RGS_LATEST_SYNC_TIMESTAMP_KEY)?;
+		kv_store.read(LATEST_RGS_SYNC_TIMESTAMP_NAMESPACE, LATEST_RGS_SYNC_TIMESTAMP_KEY)?;
 	u32::read(&mut reader).map_err(|_| {
 		std::io::Error::new(
 			std::io::ErrorKind::InvalidData,
@@ -188,21 +188,21 @@ where
 	})
 }
 
-pub(crate) fn write_rgs_latest_sync_timestamp<K: Deref, L: Deref>(
+pub(crate) fn write_latest_rgs_sync_timestamp<K: Deref, L: Deref>(
 	updated_timestamp: u32, kv_store: K, logger: L,
 ) -> Result<(), Error>
 where
 	K::Target: KVStore,
 	L::Target: Logger,
 {
 	let mut writer = kv_store
-		.write(RGS_LATEST_SYNC_TIMESTAMP_NAMESPACE, RGS_LATEST_SYNC_TIMESTAMP_KEY)
+		.write(LATEST_RGS_SYNC_TIMESTAMP_NAMESPACE, LATEST_RGS_SYNC_TIMESTAMP_KEY)
 		.map_err(|e| {
 			log_error!(
 				logger,
 				"Getting writer for key {}/{} failed due to: {}",
-				RGS_LATEST_SYNC_TIMESTAMP_NAMESPACE,
-				RGS_LATEST_SYNC_TIMESTAMP_KEY,
+				LATEST_RGS_SYNC_TIMESTAMP_NAMESPACE,
+				LATEST_RGS_SYNC_TIMESTAMP_KEY,
 				e
 			);
 			Error::PersistenceFailed
@@ -211,8 +211,8 @@ where
 		log_error!(
 			logger,
 			"Writing data to key {}/{} failed due to: {}",
-			RGS_LATEST_SYNC_TIMESTAMP_NAMESPACE,
-			RGS_LATEST_SYNC_TIMESTAMP_KEY,
+			LATEST_RGS_SYNC_TIMESTAMP_NAMESPACE,
+			LATEST_RGS_SYNC_TIMESTAMP_KEY,
 			e
 		);
 		Error::PersistenceFailed
@@ -221,8 +221,13 @@ where
 		log_error!(
 			logger,
 			"Committing data to key {}/{} failed due to: {}",
-			RGS_LATEST_SYNC_TIMESTAMP_NAMESPACE,
-			RGS_LATEST_SYNC_TIMESTAMP_KEY,
+			LATEST_RGS_SYNC_TIMESTAMP_NAMESPACE,
+			LATEST_RGS_SYNC_TIMESTAMP_KEY,
+			e
+		);
+		Error::PersistenceFailed
+	})
+}
 			e
 		);
 		Error::PersistenceFailed

src/lib.rs

@@ -172,6 +172,9 @@ const LDK_PAYMENT_RETRY_TIMEOUT: Duration = Duration::from_secs(10);
 // The time in-between peer reconnection attempts.
 const PEER_RECONNECTION_INTERVAL: Duration = Duration::from_secs(10);
 
+// The time in-between RGS sync attempts.
+const RGS_SYNC_INTERVAL: Duration = Duration::from_secs(60 * 60);
+
 // The time in-between node announcement broadcast attempts.
 const NODE_ANN_BCAST_INTERVAL: Duration = Duration::from_secs(60 * 60);
 
@@ -550,7 +553,7 @@ impl Builder {
 				));
 
 				// Reset the RGS sync timestamp in case we somehow switch gossip sources
-				io::utils::write_rgs_latest_sync_timestamp(
+				io::utils::write_latest_rgs_sync_timestamp(
 					0,
 					Arc::clone(&kv_store),
 					Arc::clone(&logger),
@@ -560,7 +563,7 @@ impl Builder {
 			}
 			GossipSourceConfig::RapidGossipSync(rgs_server) => {
 				let latest_sync_timestamp =
-					io::utils::read_rgs_latest_sync_timestamp(Arc::clone(&kv_store)).unwrap_or(0);
+					io::utils::read_latest_rgs_sync_timestamp(Arc::clone(&kv_store)).unwrap_or(0);
 				Arc::new(GossipSource::new_rgs(
 					rgs_server.clone(),
 					latest_sync_timestamp,
@@ -756,38 +759,39 @@ impl Node {
 			let gossip_source = Arc::clone(&self.gossip_source);
 			let gossip_sync_store = Arc::clone(&self.kv_store);
 			let gossip_sync_logger = Arc::clone(&self.logger);
-			let stop_gossip_sync = Arc::clone(&stop_running);
+			let mut stop_gossip_sync = self.stop_receiver.clone();
 			runtime.spawn(async move {
+				let mut interval = tokio::time::interval(RGS_SYNC_INTERVAL);
 				loop {
-					let gossip_sync_logger = Arc::clone(&gossip_sync_logger);
-					let stop_gossip_sync = Arc::clone(&stop_gossip_sync);
-					if stop_gossip_sync.load(Ordering::Acquire) {
-						return;
-					}
-
-					let now = Instant::now();
-					match gossip_source.update_rgs_snapshot().await {
-						Ok(updated_timestamp) => {
-							log_info!(
-								gossip_sync_logger,
-								"Background sync of RGS gossip data finished in {}ms.",
-								now.elapsed().as_millis()
-							);
-							io::utils::write_rgs_latest_sync_timestamp(
-								updated_timestamp,
-								Arc::clone(&gossip_sync_store),
-								Arc::clone(&gossip_sync_logger),
-							)
-							.expect("Persistence failed");
+					tokio::select! {
+						_ = stop_gossip_sync.changed() => {
+							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) => {
+									log_info!(
+										gossip_sync_logger,
+										"Background sync of RGS gossip data finished in {}ms.",
+										now.elapsed().as_millis()
+										);
+									io::utils::write_latest_rgs_sync_timestamp(
+										updated_timestamp,
+										Arc::clone(&gossip_sync_store),
+										Arc::clone(&gossip_sync_logger),
+										)
+										.expect("Persistence failed");
+								}
+								Err(e) => log_error!(
+									gossip_sync_logger,
+									"Background sync of RGS gossip data failed: {}",
+									e
+									),
+							}
 						}
-						Err(e) => log_error!(
-							gossip_sync_logger,
-							"Background sync of RGS gossip data failed: {}",
-							e
-						),
 					}
-
-					tokio::time::sleep(Duration::from_secs(60 * 60)).await;
 				}
 			});
 		}