Persist latest_node_ann_broadcast_timestamp

We remember when we last broadcasted a node announcement and only
re-announce if sufficient time has passed, we have at least one public
channel, and we have some connected peers to gossip to.

Author: tnull

src/io/mod.rs

@@ -27,8 +27,12 @@ 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";
+
+/// The last time we broadcast a node announcement will be persisted under this key.
+pub(crate) const LATEST_NODE_ANN_BCAST_TIMSTAMP_NAMESPACE: &str = "";
+pub(crate) const LATEST_NODE_ANN_BCAST_TIMSTAMP_KEY: &str = "latest_node_ann_bcast_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,65 @@ 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
+	})
+}
+
+pub(crate) fn read_latest_node_ann_bcast_timestamp<K: Deref>(
+	kv_store: K,
+) -> Result<u64, std::io::Error>
+where
+	K::Target: KVStore,
+{
+	let mut reader = kv_store
+		.read(LATEST_NODE_ANN_BCAST_TIMSTAMP_NAMESPACE, LATEST_NODE_ANN_BCAST_TIMSTAMP_KEY)?;
+	u64::read(&mut reader).map_err(|_| {
+		std::io::Error::new(
+			std::io::ErrorKind::InvalidData,
+			"Failed to deserialize latest node announcment broadcast timestamp",
+		)
+	})
+}
+
+pub(crate) fn write_latest_node_ann_bcast_timestamp<K: Deref, L: Deref>(
+	updated_timestamp: u64, kv_store: K, logger: L,
+) -> Result<(), Error>
+where
+	K::Target: KVStore,
+	L::Target: Logger,
+{
+	let mut writer = kv_store
+		.write(LATEST_NODE_ANN_BCAST_TIMSTAMP_NAMESPACE, NETWORK_GRAPH_PERSISTENCE_KEY)
+		.map_err(|e| {
+			log_error!(
+				logger,
+				"Getting writer for key {}/{} failed due to: {}",
+				LATEST_NODE_ANN_BCAST_TIMSTAMP_NAMESPACE,
+				LATEST_NODE_ANN_BCAST_TIMSTAMP_KEY,
+				e
+			);
+			Error::PersistenceFailed
+		})?;
+	updated_timestamp.write(&mut writer).map_err(|e| {
+		log_error!(
+			logger,
+			"Writing data to key {}/{} failed due to: {}",
+			LATEST_NODE_ANN_BCAST_TIMSTAMP_NAMESPACE,
+			LATEST_NODE_ANN_BCAST_TIMSTAMP_KEY,
+			e
+		);
+		Error::PersistenceFailed
+	})?;
+	writer.commit().map_err(|e| {
+		log_error!(
+			logger,
+			"Committing data to key {}/{} failed due to: {}",
+			LATEST_NODE_ANN_BCAST_TIMSTAMP_NAMESPACE,
+			LATEST_NODE_ANN_BCAST_TIMSTAMP_KEY,
 			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;
 				}
 			});
 		}
@@ -906,30 +910,51 @@ impl Node {
 		let bcast_cm = Arc::clone(&self.channel_manager);
 		let bcast_pm = Arc::clone(&self.peer_manager);
 		let bcast_config = Arc::clone(&self.config);
+		let bcast_store = Arc::clone(&self.kv_store);
+		let bcast_logger = Arc::clone(&self.logger);
 		let mut stop_bcast = self.stop_receiver.clone();
 		runtime.spawn(async move {
-			let mut interval = tokio::time::interval(NODE_ANN_BCAST_INTERVAL);
+			// We check every 30 secs whether our last broadcast is NODE_ANN_BCAST_INTERVAL away.
+			let mut interval = tokio::time::interval(Duration::from_secs(30));
 			loop {
 				tokio::select! {
 						_ = stop_bcast.changed() => {
 							return;
 						}
-						_ = interval.tick(), if bcast_cm.list_channels().iter().any(|chan| chan.is_public) => {
-							while bcast_pm.get_peer_node_ids().is_empty() {
-								// Sleep a bit and retry if we don't have any peers yet.
-								tokio::time::sleep(Duration::from_secs(5)).await;
-
-								// Check back if we need to stop.
-								match stop_bcast.has_changed() {
-									Ok(false) => {},
-									Ok(true) => return,
-									Err(_) => return,
+						_ = interval.tick() => {
+							let skip_broadcast = match io::utils::read_latest_node_ann_bcast_timestamp(Arc::clone(&bcast_store)) {
+								Ok(latest_bcast_time_secs) => {
+									// Skip if the time hasn't elapsed yet.
+									let next_bcast_unix_time = SystemTime::UNIX_EPOCH + Duration::from_secs(latest_bcast_time_secs) + NODE_ANN_BCAST_INTERVAL;
+									next_bcast_unix_time.elapsed().is_err()
+								}
+								Err(_) => {
+									// Don't skip if we haven't broadcasted before.
+									false
 								}
+							};
+
+							if skip_broadcast {
+								continue;
+							}
+
+							if bcast_cm.list_channels().iter().any(|chan| chan.is_public) {
+								// Skip if we don't have any public channels.
+								continue;
+							}
+
+							if bcast_pm.get_peer_node_ids().is_empty() {
+								// Skip if we don't have any connected peers to gossip to.
+								continue;
 							}
 
 							let addresses =
 								bcast_config.listening_address.iter().cloned().map(|a| a.0).collect();
 							bcast_pm.broadcast_node_announcement([0; 3], [0; 32], addresses);
+
+							let unix_time_secs = SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap().as_secs();
+							io::utils::write_latest_node_ann_bcast_timestamp(unix_time_secs, Arc::clone(&bcast_store), Arc::clone(&bcast_logger))
+								.expect("Persistence failed");
 						}
 				}
 			}