X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=src%2Fdownloader.rs;h=49e3019b858f362b0192ffd409835950ca6826db;hb=529c6666501812eb7b23778f02b6a61940634c88;hp=6bfaa14903b8e5f73e0154648f4bafa9f55cb3c4;hpb=63d73e758240c9d22c1d57c16aedf99cde89cd5e;p=rapid-gossip-sync-server diff --git a/src/downloader.rs b/src/downloader.rs index 6bfaa14..49e3019 100644 --- a/src/downloader.rs +++ b/src/downloader.rs @@ -1,17 +1,20 @@ +use std::ops::Deref; use std::sync::{Arc, RwLock}; use bitcoin::secp256k1::PublicKey; +use lightning::events::{MessageSendEvent, MessageSendEventsProvider}; +use lightning::ln::features::{InitFeatures, NodeFeatures}; use lightning::ln::msgs::{ChannelAnnouncement, ChannelUpdate, Init, LightningError, NodeAnnouncement, QueryChannelRange, QueryShortChannelIds, ReplyChannelRange, ReplyShortChannelIdsEnd, RoutingMessageHandler}; -use lightning::routing::gossip::{NetworkGraph, P2PGossipSync}; -use lightning::util::events::{MessageSendEvent, MessageSendEventsProvider}; +use lightning::routing::gossip::{NetworkGraph, NodeId, P2PGossipSync}; +use lightning::util::logger::Logger; use tokio::sync::mpsc; use tokio::sync::mpsc::error::TrySendError; -use crate::TestLogger; -use crate::types::{GossipMessage, GossipChainAccess}; +use crate::types::{GossipMessage, GossipChainAccess, GossipPeerManager}; use crate::verifier::ChainVerifier; pub(crate) struct GossipCounter { + pub(crate) node_announcements: u64, pub(crate) channel_announcements: u64, pub(crate) channel_updates: u64, pub(crate) channel_updates_without_htlc_max_msats: u64, @@ -21,6 +24,7 @@ pub(crate) struct GossipCounter { impl GossipCounter { pub(crate) fn new() -> Self { Self { + node_announcements: 0, channel_announcements: 0, channel_updates: 0, channel_updates_without_htlc_max_msats: 0, @@ -29,61 +33,64 @@ impl GossipCounter { } } -pub(crate) struct GossipRouter { - native_router: P2PGossipSync>, GossipChainAccess, TestLogger>, +pub(crate) struct GossipRouter where L::Target: Logger { + native_router: P2PGossipSync>, GossipChainAccess, L>, pub(crate) counter: RwLock, sender: mpsc::Sender, + verifier: Arc>, + outbound_gossiper: Arc>, GossipChainAccess, L>>, } -impl GossipRouter { - pub(crate) fn new(network_graph: Arc>, sender: mpsc::Sender) -> Self { +impl GossipRouter where L::Target: Logger { + pub(crate) fn new(network_graph: Arc>, sender: mpsc::Sender, logger: L) -> Self { + let outbound_gossiper = Arc::new(P2PGossipSync::new(Arc::clone(&network_graph), None, logger.clone())); + let verifier = Arc::new(ChainVerifier::new(Arc::clone(&network_graph), Arc::clone(&outbound_gossiper), logger.clone())); Self { - native_router: P2PGossipSync::new(network_graph, Some(Arc::new(ChainVerifier::new())), TestLogger::new()), + native_router: P2PGossipSync::new(network_graph, Some(Arc::clone(&verifier)), logger.clone()), + outbound_gossiper, counter: RwLock::new(GossipCounter::new()), - sender + sender, + verifier } } -} -impl MessageSendEventsProvider for GossipRouter { - fn get_and_clear_pending_msg_events(&self) -> Vec { - self.native_router.get_and_clear_pending_msg_events() + pub(crate) fn set_pm(&self, peer_handler: GossipPeerManager) { + self.verifier.set_ph(peer_handler); } -} -impl RoutingMessageHandler for GossipRouter { - fn handle_node_announcement(&self, msg: &NodeAnnouncement) -> Result { - self.native_router.handle_node_announcement(msg) - } + fn new_channel_announcement(&self, msg: ChannelAnnouncement) { + { + let mut counter = self.counter.write().unwrap(); + counter.channel_announcements += 1; + } - fn handle_channel_announcement(&self, msg: &ChannelAnnouncement) -> Result { - let mut counter = self.counter.write().unwrap(); + let gossip_message = GossipMessage::ChannelAnnouncement(msg, None); + if let Err(err) = self.sender.try_send(gossip_message) { + let gossip_message = match err { TrySendError::Full(msg)|TrySendError::Closed(msg) => msg }; + tokio::task::block_in_place(move || { tokio::runtime::Handle::current().block_on(async move { + self.sender.send(gossip_message).await.unwrap(); + })}); + } + } - let output_value = self.native_router.handle_channel_announcement(msg).map_err(|error| { - if error.err.contains("didn't match on-chain script") { - counter.channel_announcements_with_mismatched_scripts += 1; - } - error - })?; + fn new_node_announcement(&self, msg: NodeAnnouncement) { + { + let mut counter = self.counter.write().unwrap(); + counter.node_announcements += 1; + } - counter.channel_announcements += 1; - let gossip_message = GossipMessage::ChannelAnnouncement(msg.clone()); + let gossip_message = GossipMessage::NodeAnnouncement(msg, None); if let Err(err) = self.sender.try_send(gossip_message) { let gossip_message = match err { TrySendError::Full(msg)|TrySendError::Closed(msg) => msg }; tokio::task::block_in_place(move || { tokio::runtime::Handle::current().block_on(async move { self.sender.send(gossip_message).await.unwrap(); })}); } - - Ok(output_value) } - fn handle_channel_update(&self, msg: &ChannelUpdate) -> Result { - let output_value = self.native_router.handle_channel_update(msg)?; - - let mut counter = self.counter.write().unwrap(); - counter.channel_updates += 1; - let gossip_message = GossipMessage::ChannelUpdate(msg.clone()); + fn new_channel_update(&self, msg: ChannelUpdate) { + self.counter.write().unwrap().channel_updates += 1; + let gossip_message = GossipMessage::ChannelUpdate(msg, None); if let Err(err) = self.sender.try_send(gossip_message) { let gossip_message = match err { TrySendError::Full(msg)|TrySendError::Closed(msg) => msg }; @@ -91,20 +98,63 @@ impl RoutingMessageHandler for GossipRouter { self.sender.send(gossip_message).await.unwrap(); })}); } + } +} + +impl MessageSendEventsProvider for GossipRouter where L::Target: Logger { + fn get_and_clear_pending_msg_events(&self) -> Vec { + let gossip_evs = self.outbound_gossiper.get_and_clear_pending_msg_events(); + for ev in gossip_evs { + match ev { + MessageSendEvent::BroadcastChannelAnnouncement { msg, .. } => { + self.new_channel_announcement(msg); + }, + MessageSendEvent::BroadcastNodeAnnouncement { msg } => { + self.new_node_announcement(msg); + }, + MessageSendEvent::BroadcastChannelUpdate { msg } => { + self.new_channel_update(msg); + }, + _ => { unreachable!() }, + } + } + self.native_router.get_and_clear_pending_msg_events() + } +} - Ok(output_value) +impl RoutingMessageHandler for GossipRouter where L::Target: Logger { + fn handle_node_announcement(&self, msg: &NodeAnnouncement) -> Result { + let res = self.native_router.handle_node_announcement(msg)?; + self.new_node_announcement(msg.clone()); + Ok(res) } - fn get_next_channel_announcements(&self, starting_point: u64, batch_amount: u8) -> Vec<(ChannelAnnouncement, Option, Option)> { - self.native_router.get_next_channel_announcements(starting_point, batch_amount) + fn handle_channel_announcement(&self, msg: &ChannelAnnouncement) -> Result { + let res = self.native_router.handle_channel_announcement(msg)?; + self.new_channel_announcement(msg.clone()); + Ok(res) } - fn get_next_node_announcements(&self, starting_point: Option<&PublicKey>, batch_amount: u8) -> Vec { - self.native_router.get_next_node_announcements(starting_point, batch_amount) + fn handle_channel_update(&self, msg: &ChannelUpdate) -> Result { + let res = self.native_router.handle_channel_update(msg)?; + self.new_channel_update(msg.clone()); + Ok(res) } - fn peer_connected(&self, their_node_id: &PublicKey, init: &Init) { - self.native_router.peer_connected(their_node_id, init) + fn processing_queue_high(&self) -> bool { + self.native_router.processing_queue_high() + } + + fn get_next_channel_announcement(&self, starting_point: u64) -> Option<(ChannelAnnouncement, Option, Option)> { + self.native_router.get_next_channel_announcement(starting_point) + } + + fn get_next_node_announcement(&self, starting_point: Option<&NodeId>) -> Option { + self.native_router.get_next_node_announcement(starting_point) + } + + fn peer_connected(&self, their_node_id: &PublicKey, init: &Init, inbound: bool) -> Result<(), ()> { + self.native_router.peer_connected(their_node_id, init, inbound) } fn handle_reply_channel_range(&self, their_node_id: &PublicKey, msg: ReplyChannelRange) -> Result<(), LightningError> { @@ -122,4 +172,12 @@ impl RoutingMessageHandler for GossipRouter { fn handle_query_short_channel_ids(&self, their_node_id: &PublicKey, msg: QueryShortChannelIds) -> Result<(), LightningError> { self.native_router.handle_query_short_channel_ids(their_node_id, msg) } + + fn provided_init_features(&self, their_node_id: &PublicKey) -> InitFeatures { + self.native_router.provided_init_features(their_node_id) + } + + fn provided_node_features(&self) -> NodeFeatures { + self.native_router.provided_node_features() + } }