X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=lightning%2Fsrc%2Fln%2Frouter.rs;h=04ba31a764a2fac55b71b0b2e0b97d8b94288b1a;hb=refs%2Fheads%2F2020-04-relay-no-node;hp=0025656087bce2228a7376f2af906a5e0c131e7d;hpb=912f8774822892d6e371f9a4b1ba7a2553dbb5e7;p=rust-lightning diff --git a/lightning/src/ln/router.rs b/lightning/src/ln/router.rs index 00256560..04ba31a7 100644 --- a/lightning/src/ln/router.rs +++ b/lightning/src/ln/router.rs @@ -22,6 +22,7 @@ use util::logger::Logger; use std::cmp; use std::sync::{RwLock,Arc}; +use std::sync::atomic::{AtomicUsize, Ordering}; use std::collections::{HashMap,BinaryHeap,BTreeMap}; use std::collections::btree_map::Entry as BtreeEntry; use std; @@ -70,8 +71,8 @@ impl Writeable for Route { } } -impl Readable for Route { - fn read(reader: &mut R) -> Result { +impl Readable for Route { + fn read(reader: &mut R) -> Result { let hops_count: u8 = Readable::read(reader)?; let mut hops = Vec::with_capacity(hops_count as usize); for _ in 0..hops_count { @@ -155,7 +156,10 @@ struct NodeInfo { lowest_inbound_channel_fee_proportional_millionths: u32, features: NodeFeatures, - last_update: u32, + /// Unlike for channels, we may have a NodeInfo entry before having received a node_update. + /// Thus, we have to be able to capture "no update has been received", which we do with an + /// Option here. + last_update: Option, rgb: [u8; 3], alias: [u8; 32], addresses: Vec, @@ -166,7 +170,7 @@ struct NodeInfo { impl std::fmt::Display for NodeInfo { fn fmt(&self, f: &mut std::fmt::Formatter) -> Result<(), std::fmt::Error> { - write!(f, "features: {}, last_update: {}, lowest_inbound_channel_fee_base_msat: {}, lowest_inbound_channel_fee_proportional_millionths: {}, channels: {:?}", log_bytes!(self.features.encode()), self.last_update, self.lowest_inbound_channel_fee_base_msat, self.lowest_inbound_channel_fee_proportional_millionths, &self.channels[..])?; + write!(f, "features: {}, last_update: {:?}, lowest_inbound_channel_fee_base_msat: {}, lowest_inbound_channel_fee_proportional_millionths: {}, channels: {:?}", log_bytes!(self.features.encode()), self.last_update, self.lowest_inbound_channel_fee_base_msat, self.lowest_inbound_channel_fee_proportional_millionths, &self.channels[..])?; Ok(()) } } @@ -194,8 +198,8 @@ impl Writeable for NodeInfo { const MAX_ALLOC_SIZE: u64 = 64*1024; -impl Readable for NodeInfo { - fn read(reader: &mut R) -> Result { +impl Readable for NodeInfo { + fn read(reader: &mut R) -> Result { let channels_count: u64 = Readable::read(reader)?; let mut channels = Vec::with_capacity(cmp::min(channels_count, MAX_ALLOC_SIZE / 8) as usize); for _ in 0..channels_count { @@ -260,8 +264,8 @@ impl Writeable for NetworkMap { } } -impl Readable for NetworkMap { - fn read(reader: &mut R) -> Result { +impl Readable for NetworkMap { + fn read(reader: &mut R) -> Result { let channels_count: u64 = Readable::read(reader)?; let mut channels = BTreeMap::new(); for _ in 0..channels_count { @@ -347,6 +351,7 @@ pub struct RouteHint { pub struct Router { secp_ctx: Secp256k1, network_map: RwLock, + full_syncs_requested: AtomicUsize, chain_monitor: Arc, logger: Arc, } @@ -379,8 +384,8 @@ pub struct RouterReadArgs { pub logger: Arc, } -impl ReadableArgs for Router { - fn read(reader: &mut R, args: RouterReadArgs) -> Result { +impl ReadableArgs for Router { + fn read(reader: &mut R, args: RouterReadArgs) -> Result { let _ver: u8 = Readable::read(reader)?; let min_ver: u8 = Readable::read(reader)?; if min_ver > SERIALIZATION_VERSION { @@ -390,6 +395,7 @@ impl ReadableArgs for Router { Ok(Router { secp_ctx: Secp256k1::verification_only(), network_map: RwLock::new(network_map), + full_syncs_requested: AtomicUsize::new(0), chain_monitor: args.chain_monitor, logger: args.logger, }) @@ -406,6 +412,7 @@ macro_rules! secp_verify_sig { } impl RoutingMessageHandler for Router { + fn handle_node_announcement(&self, msg: &msgs::NodeAnnouncement) -> Result { let msg_hash = hash_to_message!(&Sha256dHash::hash(&msg.contents.encode()[..])[..]); secp_verify_sig!(self.secp_ctx, &msg_hash, &msg.signature, &msg.contents.node_id); @@ -414,12 +421,15 @@ impl RoutingMessageHandler for Router { match network.nodes.get_mut(&msg.contents.node_id) { None => Err(LightningError{err: "No existing channels for node_announcement", action: ErrorAction::IgnoreError}), Some(node) => { - if node.last_update >= msg.contents.timestamp { - return Err(LightningError{err: "Update older than last processed update", action: ErrorAction::IgnoreError}); + match node.last_update { + Some(last_update) => if last_update >= msg.contents.timestamp { + return Err(LightningError{err: "Update older than last processed update", action: ErrorAction::IgnoreError}); + }, + None => {}, } node.features = msg.contents.features.clone(); - node.last_update = msg.contents.timestamp; + node.last_update = Some(msg.contents.timestamp); node.rgb = msg.contents.rgb; node.alias = msg.contents.alias; node.addresses = msg.contents.addresses.clone(); @@ -535,7 +545,7 @@ impl RoutingMessageHandler for Router { lowest_inbound_channel_fee_base_msat: u32::max_value(), lowest_inbound_channel_fee_proportional_millionths: u32::max_value(), features: NodeFeatures::empty(), - last_update: 0, + last_update: None, rgb: [0; 3], alias: [0; 32], addresses: Vec::new(), @@ -549,6 +559,7 @@ impl RoutingMessageHandler for Router { add_channel_to_node!(msg.contents.node_id_1); add_channel_to_node!(msg.contents.node_id_2); + log_trace!(self, "Added channel_announcement for {}{}", msg.contents.short_channel_id, if !should_relay { " with excess uninterpreted data!" } else { "" }); Ok(should_relay) } @@ -653,19 +664,16 @@ impl RoutingMessageHandler for Router { Ok(msg.contents.excess_data.is_empty()) } - - fn get_next_channel_announcements(&self, starting_point: u64, batch_amount: u8) -> Vec<(msgs::ChannelAnnouncement, msgs::ChannelUpdate,msgs::ChannelUpdate)> { + fn get_next_channel_announcements(&self, starting_point: u64, batch_amount: u8) -> Vec<(msgs::ChannelAnnouncement, Option, Option)> { let mut result = Vec::with_capacity(batch_amount as usize); let network = self.network_map.read().unwrap(); let mut iter = network.channels.range(starting_point..); while result.len() < batch_amount as usize { if let Some((_, ref chan)) = iter.next() { - if chan.announcement_message.is_some() && - chan.one_to_two.last_update_message.is_some() && - chan.two_to_one.last_update_message.is_some() { + if chan.announcement_message.is_some() { result.push((chan.announcement_message.clone().unwrap(), - chan.one_to_two.last_update_message.clone().unwrap(), - chan.two_to_one.last_update_message.clone().unwrap())); + chan.one_to_two.last_update_message.clone(), + chan.two_to_one.last_update_message.clone())); } else { // TODO: We may end up sending un-announced channel_updates if we are sending // initial sync data while receiving announce/updates for this channel. @@ -698,6 +706,17 @@ impl RoutingMessageHandler for Router { } result } + + fn should_request_full_sync(&self, _node_id: &PublicKey) -> bool { + //TODO: Determine whether to request a full sync based on the network map. + const FULL_SYNCS_TO_REQUEST: usize = 5; + if self.full_syncs_requested.load(Ordering::Acquire) < FULL_SYNCS_TO_REQUEST { + self.full_syncs_requested.fetch_add(1, Ordering::AcqRel); + true + } else { + false + } + } } #[derive(Eq, PartialEq)] @@ -737,7 +756,7 @@ impl Router { lowest_inbound_channel_fee_base_msat: u32::max_value(), lowest_inbound_channel_fee_proportional_millionths: u32::max_value(), features: NodeFeatures::empty(), - last_update: 0, + last_update: None, rgb: [0; 3], alias: [0; 32], addresses: Vec::new(), @@ -750,6 +769,7 @@ impl Router { our_node_id: our_pubkey, nodes: nodes, }), + full_syncs_requested: AtomicUsize::new(0), chain_monitor, logger, } @@ -1035,7 +1055,7 @@ mod tests { use ln::channelmanager; use ln::router::{Router,NodeInfo,NetworkMap,ChannelInfo,DirectionalChannelInfo,RouteHint}; use ln::features::{ChannelFeatures, InitFeatures, NodeFeatures}; - use ln::msgs::{LightningError, ErrorAction}; + use ln::msgs::{ErrorAction, LightningError, RoutingMessageHandler}; use util::test_utils; use util::test_utils::TestVecWriter; use util::logger::Logger; @@ -1048,17 +1068,23 @@ mod tests { use hex; use secp256k1::key::{PublicKey,SecretKey}; + use secp256k1::All; use secp256k1::Secp256k1; use std::sync::Arc; - #[test] - fn route_test() { + fn create_router() -> (Secp256k1, PublicKey, Router) { let secp_ctx = Secp256k1::new(); let our_id = PublicKey::from_secret_key(&secp_ctx, &SecretKey::from_slice(&hex::decode("0101010101010101010101010101010101010101010101010101010101010101").unwrap()[..]).unwrap()); let logger: Arc = Arc::new(test_utils::TestLogger::new()); let chain_monitor = Arc::new(chaininterface::ChainWatchInterfaceUtil::new(Network::Testnet, Arc::clone(&logger))); let router = Router::new(our_id, chain_monitor, Arc::clone(&logger)); + (secp_ctx, our_id, router) + } + + #[test] + fn route_test() { + let (secp_ctx, our_id, router) = create_router(); // Build network from our_id to node8: // @@ -1153,7 +1179,7 @@ mod tests { lowest_inbound_channel_fee_base_msat: 100, lowest_inbound_channel_fee_proportional_millionths: 0, features: NodeFeatures::from_le_bytes(id_to_feature_flags!(1)), - last_update: 1, + last_update: Some(1), rgb: [0; 3], alias: [0; 32], addresses: Vec::new(), @@ -1187,7 +1213,7 @@ mod tests { lowest_inbound_channel_fee_base_msat: 0, lowest_inbound_channel_fee_proportional_millionths: 0, features: NodeFeatures::from_le_bytes(id_to_feature_flags!(2)), - last_update: 1, + last_update: Some(1), rgb: [0; 3], alias: [0; 32], addresses: Vec::new(), @@ -1221,7 +1247,7 @@ mod tests { lowest_inbound_channel_fee_base_msat: 0, lowest_inbound_channel_fee_proportional_millionths: 0, features: NodeFeatures::from_le_bytes(id_to_feature_flags!(8)), - last_update: 1, + last_update: Some(1), rgb: [0; 3], alias: [0; 32], addresses: Vec::new(), @@ -1261,7 +1287,7 @@ mod tests { lowest_inbound_channel_fee_base_msat: 0, lowest_inbound_channel_fee_proportional_millionths: 0, features: NodeFeatures::from_le_bytes(id_to_feature_flags!(3)), - last_update: 1, + last_update: Some(1), rgb: [0; 3], alias: [0; 32], addresses: Vec::new(), @@ -1341,7 +1367,7 @@ mod tests { lowest_inbound_channel_fee_base_msat: 0, lowest_inbound_channel_fee_proportional_millionths: 0, features: NodeFeatures::from_le_bytes(id_to_feature_flags!(4)), - last_update: 1, + last_update: Some(1), rgb: [0; 3], alias: [0; 32], addresses: Vec::new(), @@ -1375,7 +1401,7 @@ mod tests { lowest_inbound_channel_fee_base_msat: 0, lowest_inbound_channel_fee_proportional_millionths: 0, features: NodeFeatures::from_le_bytes(id_to_feature_flags!(5)), - last_update: 1, + last_update: Some(1), rgb: [0; 3], alias: [0; 32], addresses: Vec::new(), @@ -1432,7 +1458,7 @@ mod tests { lowest_inbound_channel_fee_base_msat: 0, lowest_inbound_channel_fee_proportional_millionths: 0, features: NodeFeatures::from_le_bytes(id_to_feature_flags!(6)), - last_update: 1, + last_update: Some(1), rgb: [0; 3], alias: [0; 32], addresses: Vec::new(), @@ -1823,4 +1849,17 @@ mod tests { assert!(::read(&mut ::std::io::Cursor::new(&w.0)).unwrap() == *network); } } + + #[test] + fn request_full_sync_finite_times() { + let (secp_ctx, _, router) = create_router(); + let node_id = PublicKey::from_secret_key(&secp_ctx, &SecretKey::from_slice(&hex::decode("0202020202020202020202020202020202020202020202020202020202020202").unwrap()[..]).unwrap()); + + assert!(router.should_request_full_sync(&node_id)); + assert!(router.should_request_full_sync(&node_id)); + assert!(router.should_request_full_sync(&node_id)); + assert!(router.should_request_full_sync(&node_id)); + assert!(router.should_request_full_sync(&node_id)); + assert!(!router.should_request_full_sync(&node_id)); + } }