Implement GossipSource to allow RGS gossip updates

Here we implement a `GossipSource` object that allows us to configure
P2P or RGS gossip sources without the need to propagate and leak LDK
type parameters upwards. To this end, `GossipSource` wraps the
corresponding variants and implements a `RoutingMessageHandler` that is
delegating or ignoring the incoming messages.

Author: tnull

src/error.rs

@@ -37,6 +37,8 @@ pub enum Error {
 	WalletSigningFailed,
 	/// A transaction sync operation failed.
 	TxSyncFailed,
+	/// A gossip updating operation failed.
+	GossipUpdateFailed,
 }
 
 impl fmt::Display for Error {
@@ -63,6 +65,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."),
 		}
 	}
 }

src/gossip.rs

@@ -0,0 +1,244 @@
+use crate::logger::{log_error, log_info, FilesystemLogger, Logger};
+use crate::types::{GossipSync, NetworkGraph, P2PGossipSync, RapidGossipSync};
+use crate::Error;
+
+use lightning::events::MessageSendEventsProvider;
+use lightning::ln::features::{InitFeatures, NodeFeatures};
+use lightning::ln::msgs as ldk_msgs;
+use lightning::ln::msgs::RoutingMessageHandler;
+use lightning::routing::gossip::NodeId;
+use lightning::routing::utxo::UtxoLookup;
+
+use bitcoin::secp256k1::PublicKey;
+
+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, 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(0);
+		Self::RapidGossipSync { gossip_sync, server_url, latest_sync_timestamp, logger }
+	}
+
+	pub fn is_p2p(&self) -> bool {
+		if let Self::P2PNetwork { .. } = self {
+			true
+		} else {
+			false
+		}
+	}
+
+	pub fn is_rgs(&self) -> bool {
+		!self.is_p2p()
+	}
+
+	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<(), Error> {
+		match self {
+			Self::P2PNetwork { gossip_sync: _ } => Ok(()),
+			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(|_| Error::GossipUpdateFailed)?;
+
+				match response.error_for_status() {
+					Ok(res) => {
+						let update_data =
+							res.bytes().await.map_err(|_| 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);
+						log_info!(logger, "Successfully retrieved latest gossip update.");
+						Ok(())
+					}
+					Err(e) => {
+						log_error!(logger, "Failed to retrieve RGS gossip update: {}", e);
+						Err(Error::GossipUpdateFailed)
+					}
+				}
+			}
+		}
+	}
+}
+
+impl MessageSendEventsProvider for GossipSource {
+	fn get_and_clear_pending_msg_events(&self) -> Vec<lightning::events::MessageSendEvent> {
+		match self {
+			GossipSource::P2PNetwork { gossip_sync, .. } => {
+				gossip_sync.get_and_clear_pending_msg_events()
+			}
+			GossipSource::RapidGossipSync { .. } => Vec::new(),
+		}
+	}
+}
+
+impl RoutingMessageHandler for GossipSource {
+	fn handle_node_announcement(
+		&self, msg: &ldk_msgs::NodeAnnouncement,
+	) -> Result<bool, ldk_msgs::LightningError> {
+		match self {
+			GossipSource::P2PNetwork { gossip_sync, .. } => {
+				gossip_sync.handle_node_announcement(msg)
+			}
+			GossipSource::RapidGossipSync { .. } => Ok(false),
+		}
+	}
+
+	fn handle_channel_announcement(
+		&self, msg: &ldk_msgs::ChannelAnnouncement,
+	) -> Result<bool, ldk_msgs::LightningError> {
+		match self {
+			GossipSource::P2PNetwork { gossip_sync, .. } => {
+				gossip_sync.handle_channel_announcement(msg)
+			}
+			GossipSource::RapidGossipSync { .. } => Ok(false),
+		}
+	}
+
+	fn handle_channel_update(
+		&self, msg: &ldk_msgs::ChannelUpdate,
+	) -> Result<bool, ldk_msgs::LightningError> {
+		match self {
+			GossipSource::P2PNetwork { gossip_sync, .. } => gossip_sync.handle_channel_update(msg),
+			GossipSource::RapidGossipSync { .. } => Ok(false),
+		}
+	}
+
+	fn get_next_channel_announcement(
+		&self, starting_point: u64,
+	) -> Option<(
+		ldk_msgs::ChannelAnnouncement,
+		Option<ldk_msgs::ChannelUpdate>,
+		Option<ldk_msgs::ChannelUpdate>,
+	)> {
+		match self {
+			GossipSource::P2PNetwork { gossip_sync, .. } => {
+				gossip_sync.get_next_channel_announcement(starting_point)
+			}
+			GossipSource::RapidGossipSync { .. } => None,
+		}
+	}
+
+	fn get_next_node_announcement(
+		&self, starting_point: Option<&NodeId>,
+	) -> Option<ldk_msgs::NodeAnnouncement> {
+		match self {
+			GossipSource::P2PNetwork { gossip_sync, .. } => {
+				gossip_sync.get_next_node_announcement(starting_point)
+			}
+			GossipSource::RapidGossipSync { .. } => None,
+		}
+	}
+
+	fn peer_connected(
+		&self, their_node_id: &PublicKey, init: &ldk_msgs::Init, inbound: bool,
+	) -> Result<(), ()> {
+		match self {
+			GossipSource::P2PNetwork { gossip_sync, .. } => {
+				gossip_sync.peer_connected(their_node_id, init, inbound)
+			}
+			GossipSource::RapidGossipSync { .. } => Ok(()),
+		}
+	}
+
+	fn handle_reply_channel_range(
+		&self, their_node_id: &PublicKey, msg: ldk_msgs::ReplyChannelRange,
+	) -> Result<(), ldk_msgs::LightningError> {
+		match self {
+			GossipSource::P2PNetwork { gossip_sync, .. } => {
+				gossip_sync.handle_reply_channel_range(their_node_id, msg)
+			}
+			GossipSource::RapidGossipSync { .. } => Ok(()),
+		}
+	}
+
+	fn handle_reply_short_channel_ids_end(
+		&self, their_node_id: &PublicKey, msg: ldk_msgs::ReplyShortChannelIdsEnd,
+	) -> Result<(), ldk_msgs::LightningError> {
+		match self {
+			GossipSource::P2PNetwork { gossip_sync, .. } => {
+				gossip_sync.handle_reply_short_channel_ids_end(their_node_id, msg)
+			}
+			GossipSource::RapidGossipSync { .. } => Ok(()),
+		}
+	}
+
+	fn handle_query_channel_range(
+		&self, their_node_id: &PublicKey, msg: ldk_msgs::QueryChannelRange,
+	) -> Result<(), ldk_msgs::LightningError> {
+		match self {
+			GossipSource::P2PNetwork { gossip_sync, .. } => {
+				gossip_sync.handle_query_channel_range(their_node_id, msg)
+			}
+			GossipSource::RapidGossipSync { .. } => Ok(()),
+		}
+	}
+
+	fn handle_query_short_channel_ids(
+		&self, their_node_id: &PublicKey, msg: ldk_msgs::QueryShortChannelIds,
+	) -> Result<(), ldk_msgs::LightningError> {
+		match self {
+			GossipSource::P2PNetwork { gossip_sync, .. } => {
+				gossip_sync.handle_query_short_channel_ids(their_node_id, msg)
+			}
+			GossipSource::RapidGossipSync { .. } => Ok(()),
+		}
+	}
+
+	fn provided_node_features(&self) -> NodeFeatures {
+		match self {
+			GossipSource::P2PNetwork { gossip_sync, .. } => gossip_sync.provided_node_features(),
+			GossipSource::RapidGossipSync { .. } => NodeFeatures::empty(),
+		}
+	}
+	fn provided_init_features(&self, their_node_id: &PublicKey) -> InitFeatures {
+		match self {
+			GossipSource::P2PNetwork { gossip_sync, .. } => {
+				gossip_sync.provided_init_features(their_node_id)
+			}
+			GossipSource::RapidGossipSync { .. } => InitFeatures::empty(),
+		}
+	}
+
+	fn processing_queue_high(&self) -> bool {
+		match self {
+			GossipSource::P2PNetwork { gossip_sync, .. } => gossip_sync.processing_queue_high(),
+			GossipSource::RapidGossipSync { .. } => false,
+		}
+	}
+}