From a9b38dc1aecf0ebe9f84489f359cc8650dbf4d74 Mon Sep 17 00:00:00 2001 From: Slack Coder Date: Mon, 7 Nov 2022 07:50:09 -0500 Subject: [PATCH] node: let others know their address Include the node's IP address in the Node's announcements. Source it from the service's configuration which can assumed to be populated with default values when available. Signed-off-by: Slack Coder --- radicle-node/src/main.rs | 13 +++++- radicle-node/src/service.rs | 53 ++++++++-------------- radicle-node/src/service/config.rs | 3 ++ radicle-node/src/test/peer.rs | 5 ++- radicle-node/src/tests.rs | 70 +++++++++++++++++++++++------- 5 files changed, 91 insertions(+), 53 deletions(-) diff --git a/radicle-node/src/main.rs b/radicle-node/src/main.rs index 30362e43..b875fb2d 100644 --- a/radicle-node/src/main.rs +++ b/radicle-node/src/main.rs @@ -14,6 +14,7 @@ type Reactor = nakamoto_net_poll::Reactor; struct Options { connect: Vec
, listen: Vec, + external_addresses: Vec
, } impl Options { @@ -22,6 +23,7 @@ impl Options { let mut parser = lexopt::Parser::from_env(); let mut connect = Vec::new(); + let mut external_addresses = Vec::new(); let mut listen = Vec::new(); while let Some(arg) = parser.next()? { @@ -34,6 +36,10 @@ impl Options { let addr = parser.value()?.parse()?; listen.push(addr); } + Long("external-address") => { + let addr = parser.value()?.parse()?; + external_addresses.push(addr); + } Long("help") => { println!("usage: radicle-node [--connect ].."); process::exit(0); @@ -41,7 +47,11 @@ impl Options { _ => return Err(arg.unexpected()), } } - Ok(Self { connect, listen }) + Ok(Self { + connect, + listen, + external_addresses, + }) } } @@ -66,6 +76,7 @@ fn main() -> anyhow::Result<()> { let config = client::Config { service: service::Config { connect: options.connect, + external_addresses: options.external_addresses, ..service::Config::default() }, listen: options.listen, diff --git a/radicle-node/src/service.rs b/radicle-node/src/service.rs index de7e16e4..1c28b979 100644 --- a/radicle-node/src/service.rs +++ b/radicle-node/src/service.rs @@ -945,29 +945,22 @@ where } } - fn next_connections( - rng: &Rng, - sessions: &Sessions, - address_pool: &mut dyn Iterator, - ) -> Vec
{ + fn choose_addresses(&mut self) -> Vec
{ let mut initializing: Vec
= Vec::new(); let mut negotiated: HashMap = HashMap::new(); - for (_, s) in sessions.iter() { - if s.link != Link::Outbound { + for s in self.sessions.values() { + if !s.link.is_outbound() { continue; } match s.state { session::State::Initial => { initializing.push(s.addr.into()); } - session::State::Negotiated { .. } => { - let node_id = s - .node_id() - .expect("negotiated sessions must have a node ID"); - negotiated.insert(node_id, s); + session::State::Negotiated { id, .. } => { + negotiated.insert(id, s); } - _ => continue, - }; + session::State::Disconnected { .. } => {} + } } let wanted = TARGET_OUTBOUND_PEERS @@ -977,33 +970,23 @@ where return Vec::new(); } - // All nodes are considered equal - let mut address_pool: Vec<_> = address_pool + self.addresses + .entries() + .unwrap() .filter(|(node_id, s)| { !initializing.contains(&s.addr) && !negotiated.contains_key(node_id) }) .take(wanted) - .collect(); - - let mut next = Vec::new(); - loop { - if address_pool.is_empty() { - break; - } - let i = rng.usize(0..address_pool.len()); - let (_, addr) = address_pool.swap_remove(i); - next.push(addr.addr); - } - next + .map(|(_, s)| s.addr) + .collect() } fn maintain_connections(&mut self) { - let mut address_pool = self - .addresses - .entries() - .expect("address store be accessible"); - let next = Self::next_connections(&self.rng, &self.sessions, &mut address_pool); - for addr in next { + let addrs = self.choose_addresses(); + if addrs.is_empty() { + debug!("No eligible peers available to connect to"); + } + for addr in addrs { self.reactor.connect(addr.clone()); } } @@ -1267,7 +1250,7 @@ mod gossip { pub fn node(timestamp: Timestamp, config: &Config) -> NodeAnnouncement { let features = node::Features::SEED; let alias = config.alias(); - let addresses = vec![]; // TODO + let addresses = config.external_addresses.clone(); NodeAnnouncement { features, diff --git a/radicle-node/src/service/config.rs b/radicle-node/src/service/config.rs index 8017d35e..c48155bb 100644 --- a/radicle-node/src/service/config.rs +++ b/radicle-node/src/service/config.rs @@ -46,6 +46,8 @@ pub struct Config { /// Peers to connect to on startup. /// Connections to these peers will be maintained. pub connect: Vec
, + /// Specify the node's public addresses + pub external_addresses: Vec
, /// Peer-to-peer network. pub network: Network, /// Project tracking policy. @@ -62,6 +64,7 @@ impl Default for Config { fn default() -> Self { Self { connect: Vec::default(), + external_addresses: vec![], network: Network::default(), project_tracking: ProjectTracking::default(), remote_tracking: RemoteTracking::default(), diff --git a/radicle-node/src/test/peer.rs b/radicle-node/src/test/peer.rs index cfbe55f7..1e4fd635 100644 --- a/radicle-node/src/test/peer.rs +++ b/radicle-node/src/test/peer.rs @@ -75,7 +75,8 @@ where let mut rng = fastrand::Rng::new(); let signer = MockSigner::new(&mut rng); - Self::config(name, Config::default(), ip, storage, signer, rng) + let addrs = address::Book::memory().unwrap(); + Self::config(name, Config::default(), ip, storage, addrs, signer, rng) } } @@ -89,13 +90,13 @@ where config: Config, ip: impl Into, storage: S, + addrs: address::Book, signer: G, rng: fastrand::Rng, ) -> Self { let local_time = LocalTime::now(); let clock = RefClock::from(local_time); let routing = routing::Table::memory().unwrap(); - let addrs = address::Book::memory().unwrap(); let service = Service::new(config, clock, routing, storage, addrs, signer, rng.clone()); let ip = ip.into(); let local_addr = net::SocketAddr::new(ip, rng.u16(..)); diff --git a/radicle-node/src/tests.rs b/radicle-node/src/tests.rs index 9a8cf705..804014dc 100644 --- a/radicle-node/src/tests.rs +++ b/radicle-node/src/tests.rs @@ -4,6 +4,7 @@ use std::sync::Arc; use crossbeam_channel as chan; use nakamoto_net as nakamoto; +use crate::address; use crate::collections::{HashMap, HashSet}; use crate::crypto::test::signer::MockSigner; use crate::prelude::{LocalDuration, Timestamp}; @@ -169,6 +170,7 @@ fn test_persistent_peer_connect() { config, [7, 7, 7, 7], MockStorage::empty(), + address::Book::memory().unwrap(), MockSigner::new(&mut rng), rng, ); @@ -235,6 +237,7 @@ fn test_tracking() { }, [7, 7, 7, 7], MockStorage::empty(), + address::Book::memory().unwrap(), MockSigner::default(), fastrand::Rng::new(), ); @@ -428,10 +431,19 @@ fn test_refs_announcement_relay() { let bob = { let mut rng = fastrand::Rng::new(); + let addresses = address::Book::memory().unwrap(); let signer = MockSigner::new(&mut rng); let storage = fixtures::storage(tmp.path().join("bob"), &signer).unwrap(); - Peer::config("bob", Config::default(), [9, 9, 9, 9], storage, signer, rng) + Peer::config( + "bob", + Config::default(), + [9, 9, 9, 9], + storage, + addresses, + signer, + rng, + ) }; let bob_inv = bob.inventory().unwrap(); @@ -593,6 +605,7 @@ fn test_persistent_peer_reconnect() { }, [7, 7, 7, 7], MockStorage::empty(), + address::Book::memory().unwrap(), MockSigner::default(), fastrand::Rng::new(), ); @@ -648,21 +661,49 @@ fn test_persistent_peer_reconnect() { assert_matches!(alice.outbox().next(), None); } +fn add_peer(address_book: &mut A, peer: &Peer) { + let known_address = address::KnownAddress::new(peer.address(), address::Source::Peer); + address_book + .insert( + &peer.node_id(), + radicle::node::Features::NONE, + peer.name, + LocalTime::now().as_secs(), + Some(known_address), + ) + .unwrap(); +} + #[test] fn test_maintain_connections() { - let mut alice = Peer::new("alice", [7, 7, 7, 7], MockStorage::empty()); - let connected = vec![ - Peer::new("connected 1", [8, 8, 8, 1], MockStorage::empty()), - Peer::new("connected 2", [8, 8, 8, 2], MockStorage::empty()), - Peer::new("connected 3", [8, 8, 8, 3], MockStorage::empty()), + Peer::new("connected", [8, 8, 8, 1], MockStorage::empty()), + Peer::new("connected", [8, 8, 8, 2], MockStorage::empty()), + Peer::new("connected", [8, 8, 8, 3], MockStorage::empty()), ]; let mut unconnected = vec![ - Peer::new("new 1", [9, 9, 9, 1], MockStorage::empty()), - Peer::new("new 2", [9, 9, 9, 2], MockStorage::empty()), - Peer::new("new 3", [9, 9, 9, 3], MockStorage::empty()), + Peer::new("unconnected", [9, 9, 9, 1], MockStorage::empty()), + Peer::new("unconnected", [9, 9, 9, 2], MockStorage::empty()), + Peer::new("unconnected", [9, 9, 9, 3], MockStorage::empty()), ]; + let mut address_book = address::Book::memory().unwrap(); + for peer in &unconnected { + add_peer(&mut address_book, peer); + } + + let mut alice = Peer::config( + "alice", + Config { + project_tracking: ProjectTracking::Allowed(HashSet::default()), + ..Config::default() + }, + [7, 7, 7, 7], + MockStorage::empty(), + address_book, + MockSigner::default(), + fastrand::Rng::new(), + ); for peer in connected.iter() { alice.connect_to(peer); } @@ -672,14 +713,10 @@ fn test_maintain_connections() { "alice should be connected to all peers" ); - for peer in unconnected.iter() { - alice.receive(&connected[0].addr(), peer.node_announcement()); - } - for peer in connected.iter() { alice.disconnected( &peer.addr(), - &nakamoto::DisconnectReason::Protocol(DisconnectReason::User), + &nakamoto::DisconnectReason::Protocol(DisconnectReason::Error(session::Error::Timeout)), ); let addr = alice @@ -692,7 +729,10 @@ fn test_maintain_connections() { assert!(addr != peer.addr()); unconnected.retain(|p| p.addr() != addr); } - assert!(unconnected.is_empty()); + assert!( + unconnected.is_empty(), + "alice should connect to all unconnected peers" + ); } #[test]