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

@@ -26,6 +26,10 @@ pub(crate) const PEER_INFO_PERSISTENCE_KEY: &str = "peers";
 /// The payment information will be persisted under this prefix.
 pub(crate) const PAYMENT_INFO_PERSISTENCE_NAMESPACE: &str = "payments";
 
+/// 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

@@ -1,6 +1,7 @@
 use super::*;
-use crate::WALLET_KEYS_SEED_LEN;
+use crate::{Error, WALLET_KEYS_SEED_LEN};
 
+use crate::logger::log_error;
 use crate::peer_store::PeerStore;
 use crate::{EventQueue, PaymentDetails};
 
@@ -9,7 +10,7 @@ use lightning::chain::keysinterface::{EntropySource, SignerProvider};
 use lightning::routing::gossip::NetworkGraph;
 use lightning::routing::scoring::{ProbabilisticScorer, ProbabilisticScoringParameters};
 use lightning::util::logger::Logger;
-use lightning::util::ser::{Readable, ReadableArgs};
+use lightning::util::ser::{Readable, ReadableArgs, Writeable};
 
 use bitcoin::hash_types::{BlockHash, Txid};
 use bitcoin::hashes::hex::FromHex;
@@ -172,3 +173,60 @@ where
 	}
 	Ok(res)
 }
+
+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

@@ -802,30 +802,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(1));
 			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(last_bcast_time_secs) => {
+									// Skip if the time hasn't elapsed yet.
+									let next_bcast_unix_time = SystemTime::UNIX_EPOCH + Duration::from_secs(last_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");
 						}
 				}
 			}