Implement configurable GossipSource to allow RGS gossip updates

Cargo.toml

@@ -14,7 +14,7 @@ categories = ["cryptography::cryptocurrencies"]
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [lib]
-crate-type = ["staticlib", "cdylib"]
+crate-type = ["lib", "staticlib", "cdylib"]
 name = "ldk_node"
 
 [[bin]]

bindings/ldk_node.udl

@@ -77,6 +77,7 @@ enum NodeError {
 	"WalletOperationFailed",
 	"WalletSigningFailed",
 	"TxSyncFailed",
+	"GossipUpdateFailed",
 	"InvalidAddress",
 	"InvalidPublicKey",
 	"InvalidPaymentHash",

src/error.rs

@@ -29,6 +29,8 @@ pub enum Error {
 	WalletSigningFailed,
 	/// A transaction sync operation failed.
 	TxSyncFailed,
+	/// A gossip updating operation failed.
+	GossipUpdateFailed,
 	/// The given address is invalid.
 	InvalidAddress,
 	/// The given public key is invalid.
@@ -71,6 +73,7 @@ impl fmt::Display for Error {
 			Self::WalletOperationFailed => write!(f, "Failed to conduct wallet operation."),
 			Self::WalletSigningFailed => write!(f, "Failed to sign given transaction."),
 			Self::TxSyncFailed => write!(f, "Failed to sync transactions."),
+			Self::GossipUpdateFailed => write!(f, "Failed to update gossip data."),
 			Self::InvalidAddress => write!(f, "The given address is invalid."),
 			Self::InvalidPublicKey => write!(f, "The given public key is invalid."),
 			Self::InvalidPaymentHash => write!(f, "The given payment hash is invalid."),

src/gossip.rs

@@ -0,0 +1,90 @@
+use crate::logger::{log_trace, FilesystemLogger, Logger};
+use crate::types::{GossipSync, NetworkGraph, P2PGossipSync, RapidGossipSync};
+use crate::Error;
+
+use lightning::routing::utxo::UtxoLookup;
+
+use std::sync::atomic::{AtomicU32, Ordering};
+use std::sync::Arc;
+
+pub(crate) enum GossipSource {
+	P2PNetwork {
+		gossip_sync: Arc<P2PGossipSync>,
+	},
+	RapidGossipSync {
+		gossip_sync: Arc<RapidGossipSync>,
+		server_url: String,
+		latest_sync_timestamp: AtomicU32,
+		logger: Arc<FilesystemLogger>,
+	},
+}
+
+impl GossipSource {
+	pub fn new_p2p(network_graph: Arc<NetworkGraph>, logger: Arc<FilesystemLogger>) -> Self {
+		let gossip_sync = Arc::new(P2PGossipSync::new(
+			network_graph,
+			None::<Arc<dyn UtxoLookup + Send + Sync>>,
+			logger,
+		));
+		Self::P2PNetwork { gossip_sync }
+	}
+
+	pub fn new_rgs(
+		server_url: String, latest_sync_timestamp: u32, network_graph: Arc<NetworkGraph>,
+		logger: Arc<FilesystemLogger>,
+	) -> Self {
+		let gossip_sync = Arc::new(RapidGossipSync::new(network_graph, Arc::clone(&logger)));
+		let latest_sync_timestamp = AtomicU32::new(latest_sync_timestamp);
+		Self::RapidGossipSync { gossip_sync, server_url, latest_sync_timestamp, logger }
+	}
+
+	pub fn is_rgs(&self) -> bool {
+		if let Self::RapidGossipSync { .. } = self {
+			true
+		} else {
+			false
+		}
+	}
+
+	pub fn as_gossip_sync(&self) -> GossipSync {
+		match self {
+			Self::RapidGossipSync { gossip_sync, .. } => {
+				GossipSync::Rapid(Arc::clone(&gossip_sync))
+			}
+			Self::P2PNetwork { gossip_sync, .. } => GossipSync::P2P(Arc::clone(&gossip_sync)),
+		}
+	}
+
+	pub async fn update_rgs_snapshot(&self) -> Result<u32, Error> {
+		match self {
+			Self::P2PNetwork { gossip_sync: _ } => Ok(0),
+			Self::RapidGossipSync { gossip_sync, server_url, latest_sync_timestamp, logger } => {
+				let query_timestamp = latest_sync_timestamp.load(Ordering::Acquire);
+				let query_url = format!("{}/{}", server_url, query_timestamp);
+				let response = reqwest::get(query_url).await.map_err(|e| {
+					log_trace!(logger, "Failed to retrieve RGS gossip update: {}", e);
+					Error::GossipUpdateFailed
+				})?;
+
+				match response.error_for_status() {
+					Ok(res) => {
+						let update_data = res.bytes().await.map_err(|e| {
+							log_trace!(logger, "Failed to retrieve RGS gossip update: {}", e);
+							Error::GossipUpdateFailed
+						})?;
+
+						let new_latest_sync_timestamp = gossip_sync
+							.update_network_graph(&update_data)
+							.map_err(|_| Error::GossipUpdateFailed)?;
+						latest_sync_timestamp.store(new_latest_sync_timestamp, Ordering::Release);
+						Ok(new_latest_sync_timestamp)
+					}
+					Err(e) => {
+						log_trace!(logger, "Failed to retrieve RGS gossip update: {}", e);
+						Err(Error::GossipUpdateFailed)
+					}
+				}
+			}
+		}
+	}
+}

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";
 
+/// 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";
+
 /// Provides an interface that allows to store and retrieve persisted values that are associated
 /// with given keys.
 ///

src/io/utils.rs

@@ -1,15 +1,16 @@
 use super::*;
 use crate::WALLET_KEYS_SEED_LEN;
 
+use crate::logger::log_error;
 use crate::peer_store::PeerStore;
-use crate::{EventQueue, PaymentDetails};
+use crate::{Error, EventQueue, PaymentDetails};
 
 use lightning::chain::channelmonitor::ChannelMonitor;
 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,58 @@ where
 	}
 	Ok(res)
 }
+
+pub(crate) fn read_rgs_latest_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)?;
+	u32::read(&mut reader).map_err(|_| {
+		std::io::Error::new(
+			std::io::ErrorKind::InvalidData,
+			"Failed to deserialize latest RGS sync timestamp",
+		)
+	})
+}
+
+pub(crate) fn write_rgs_latest_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)
+		.map_err(|e| {
+			log_error!(
+				logger,
+				"Getting writer for key {}/{} failed due to: {}",
+				RGS_LATEST_SYNC_TIMESTAMP_NAMESPACE,
+				RGS_LATEST_SYNC_TIMESTAMP_KEY,
+				e
+			);
+			Error::PersistenceFailed
+		})?;
+	updated_timestamp.write(&mut writer).map_err(|e| {
+		log_error!(
+			logger,
+			"Writing data to key {}/{} failed due to: {}",
+			RGS_LATEST_SYNC_TIMESTAMP_NAMESPACE,
+			RGS_LATEST_SYNC_TIMESTAMP_KEY,
+			e
+		);
+		Error::PersistenceFailed
+	})?;
+	writer.commit().map_err(|e| {
+		log_error!(
+			logger,
+			"Committing data to key {}/{} failed due to: {}",
+			RGS_LATEST_SYNC_TIMESTAMP_NAMESPACE,
+			RGS_LATEST_SYNC_TIMESTAMP_KEY,
+			e
+		);
+		Error::PersistenceFailed
+	})
+}