node: Include addresses in `Command::Seeds` result
This will allow the CLI to connect to seeds on the go. We also do some refactoring of some of the code involved in fulfilling the `Seeds` command.
This commit is contained in:
parent
26e3c0fbff
commit
8a3133e137
|
|
@ -149,12 +149,12 @@ pub fn run(options: Options, ctx: impl term::Context) -> anyhow::Result<()> {
|
||||||
|
|
||||||
fn announce(rid: Id, timeout: time::Duration, mut node: Node) -> anyhow::Result<()> {
|
fn announce(rid: Id, timeout: time::Duration, mut node: Node) -> anyhow::Result<()> {
|
||||||
let seeds = node.seeds(rid)?;
|
let seeds = node.seeds(rid)?;
|
||||||
if !seeds.has_connections() {
|
let connected = seeds.connected().map(|s| s.nid).collect::<Vec<_>>();
|
||||||
|
if connected.is_empty() {
|
||||||
term::info!("Not connected to any seeds.");
|
term::info!("Not connected to any seeds.");
|
||||||
return Ok(());
|
return Ok(());
|
||||||
}
|
}
|
||||||
|
|
||||||
let connected = seeds.connected().cloned().collect::<Vec<_>>();
|
|
||||||
let mut spinner = term::spinner(format!("Syncing with {} node(s)..", connected.len()));
|
let mut spinner = term::spinner(format!("Syncing with {} node(s)..", connected.len()));
|
||||||
let result = node.announce(rid, connected, timeout, |event| match event {
|
let result = node.announce(rid, connected, timeout, |event| match event {
|
||||||
node::AnnounceEvent::Announced => {}
|
node::AnnounceEvent::Announced => {}
|
||||||
|
|
@ -210,12 +210,10 @@ pub fn fetch_all(rid: Id, node: &mut Node) -> Result<FetchResults, node::Error>
|
||||||
let seeds = node.seeds(rid)?;
|
let seeds = node.seeds(rid)?;
|
||||||
let mut results = FetchResults::default();
|
let mut results = FetchResults::default();
|
||||||
|
|
||||||
if seeds.has_connections() {
|
// Fetch from connected seeds.
|
||||||
// Fetch from all seeds.
|
for seed in seeds.connected() {
|
||||||
for seed in seeds.connected() {
|
let result = fetch_from(rid, &seed.nid, node)?;
|
||||||
let result = fetch_from(rid, seed, node)?;
|
results.push(seed.nid, result);
|
||||||
results.push(*seed, result);
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
Ok(results)
|
Ok(results)
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -636,8 +636,9 @@ fn test_clone_without_seeds() {
|
||||||
let rid = alice.project("heartwood", "Radicle Heartwood Protocol & Stack");
|
let rid = alice.project("heartwood", "Radicle Heartwood Protocol & Stack");
|
||||||
let mut alice = alice.spawn();
|
let mut alice = alice.spawn();
|
||||||
let seeds = alice.handle.seeds(rid).unwrap();
|
let seeds = alice.handle.seeds(rid).unwrap();
|
||||||
|
let connected = seeds.connected().collect::<Vec<_>>();
|
||||||
|
|
||||||
assert!(!seeds.has_connections());
|
assert!(connected.is_empty());
|
||||||
|
|
||||||
alice
|
alice
|
||||||
.rad("clone", &[rid.to_string().as_str()], working.as_path())
|
.rad("clone", &[rid.to_string().as_str()], working.as_path())
|
||||||
|
|
|
||||||
|
|
@ -17,6 +17,7 @@ use crossbeam_channel as chan;
|
||||||
use fastrand::Rng;
|
use fastrand::Rng;
|
||||||
use localtime::{LocalDuration, LocalTime};
|
use localtime::{LocalDuration, LocalTime};
|
||||||
use log::*;
|
use log::*;
|
||||||
|
use nonempty::NonEmpty;
|
||||||
|
|
||||||
use radicle::node::address;
|
use radicle::node::address;
|
||||||
use radicle::node::address::{AddressBook, KnownAddress};
|
use radicle::node::address::{AddressBook, KnownAddress};
|
||||||
|
|
@ -502,10 +503,11 @@ where
|
||||||
}
|
}
|
||||||
Command::Seeds(rid, resp) => match self.seeds(&rid) {
|
Command::Seeds(rid, resp) => match self.seeds(&rid) {
|
||||||
Ok(seeds) => {
|
Ok(seeds) => {
|
||||||
|
let (connected, disconnected) = seeds.partition();
|
||||||
debug!(
|
debug!(
|
||||||
target: "service",
|
target: "service",
|
||||||
"Found {} connected seed(s) and {} disconnected seed(s) for {}",
|
"Found {} connected seed(s) and {} disconnected seed(s) for {}",
|
||||||
seeds.connected().count(), seeds.disconnected().count(), rid
|
connected.len(), disconnected.len(), rid
|
||||||
);
|
);
|
||||||
resp.send(seeds).ok();
|
resp.send(seeds).ok();
|
||||||
}
|
}
|
||||||
|
|
@ -1322,28 +1324,25 @@ where
|
||||||
}
|
}
|
||||||
|
|
||||||
fn seeds(&self, rid: &Id) -> Result<Seeds, Error> {
|
fn seeds(&self, rid: &Id) -> Result<Seeds, Error> {
|
||||||
#[derive(Default)]
|
let seeds = match self.routing.get(rid) {
|
||||||
pub struct Stats {
|
Ok(seeds) => seeds.into_iter().fold(Seeds::default(), |mut seeds, node| {
|
||||||
connected: usize,
|
if node != self.node_id() {
|
||||||
disconnected: usize,
|
let addrs: Vec<KnownAddress> = self
|
||||||
}
|
.addresses
|
||||||
|
.get(&node)
|
||||||
|
.ok()
|
||||||
|
.flatten()
|
||||||
|
.map(|n| n.addrs)
|
||||||
|
.unwrap_or(vec![]);
|
||||||
|
|
||||||
let (_, seeds) = match self.routing.get(rid) {
|
if let Some(s) = self.sessions.get(&node) {
|
||||||
Ok(seeds) => seeds.into_iter().fold(
|
seeds.insert(Seed::new(node, addrs, Some(s.state.clone())));
|
||||||
(Stats::default(), Seeds::default()),
|
} else {
|
||||||
|(mut stats, mut seeds), node| {
|
seeds.insert(Seed::new(node, addrs, None));
|
||||||
if node != self.node_id() {
|
|
||||||
if self.sessions.is_connected(&node) {
|
|
||||||
seeds.insert(Seed::Connected(node));
|
|
||||||
stats.connected += 1;
|
|
||||||
} else if self.sessions.is_disconnected(&node) {
|
|
||||||
seeds.insert(Seed::Disconnected(node));
|
|
||||||
stats.disconnected += 1;
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
(stats, seeds)
|
}
|
||||||
},
|
seeds
|
||||||
),
|
}),
|
||||||
Err(err) => {
|
Err(err) => {
|
||||||
return Err(Error::Routing(err));
|
return Err(Error::Routing(err));
|
||||||
}
|
}
|
||||||
|
|
@ -1464,9 +1463,9 @@ where
|
||||||
for rid in missing {
|
for rid in missing {
|
||||||
match self.seeds(&rid) {
|
match self.seeds(&rid) {
|
||||||
Ok(seeds) => {
|
Ok(seeds) => {
|
||||||
if seeds.has_connections() {
|
if let Some(connected) = NonEmpty::from_vec(seeds.connected().collect()) {
|
||||||
for seed in seeds.connected() {
|
for seed in connected {
|
||||||
self.fetch(rid, seed);
|
self.fetch(rid, &seed.nid);
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
// TODO: We should make sure that this fetch is retried later, either
|
// TODO: We should make sure that this fetch is retried later, either
|
||||||
|
|
|
||||||
|
|
@ -509,7 +509,7 @@ fn push_ref(
|
||||||
/// Sync with the network.
|
/// Sync with the network.
|
||||||
fn sync(rid: Id, mut node: radicle::Node) -> Result<(), radicle::node::Error> {
|
fn sync(rid: Id, mut node: radicle::Node) -> Result<(), radicle::node::Error> {
|
||||||
let seeds = node.seeds(rid)?;
|
let seeds = node.seeds(rid)?;
|
||||||
let connected = seeds.connected().cloned().collect::<Vec<_>>();
|
let connected = seeds.connected().map(|s| s.nid).collect::<Vec<_>>();
|
||||||
|
|
||||||
if connected.is_empty() {
|
if connected.is_empty() {
|
||||||
eprintln!("Not connected to any seeds.");
|
eprintln!("Not connected to any seeds.");
|
||||||
|
|
|
||||||
|
|
@ -6,7 +6,7 @@ pub mod events;
|
||||||
pub mod routing;
|
pub mod routing;
|
||||||
pub mod tracking;
|
pub mod tracking;
|
||||||
|
|
||||||
use std::collections::{BTreeSet, HashMap, HashSet};
|
use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
|
||||||
use std::io::{BufRead, BufReader};
|
use std::io::{BufRead, BufReader};
|
||||||
use std::ops::Deref;
|
use std::ops::Deref;
|
||||||
use std::os::unix::net::UnixStream;
|
use std::os::unix::net::UnixStream;
|
||||||
|
|
@ -25,6 +25,7 @@ use crate::crypto::PublicKey;
|
||||||
use crate::identity::Id;
|
use crate::identity::Id;
|
||||||
use crate::storage::RefUpdate;
|
use crate::storage::RefUpdate;
|
||||||
|
|
||||||
|
pub use address::KnownAddress;
|
||||||
pub use config::Config;
|
pub use config::Config;
|
||||||
pub use cyphernet::addr::PeerAddr;
|
pub use cyphernet::addr::PeerAddr;
|
||||||
pub use events::{Event, Events};
|
pub use events::{Event, Events};
|
||||||
|
|
@ -359,49 +360,50 @@ pub struct Session {
|
||||||
pub state: State,
|
pub state: State,
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, serde::Serialize, serde::Deserialize)]
|
#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
|
||||||
#[serde(rename_all = "kebab-case")]
|
#[serde(rename_all = "kebab-case")]
|
||||||
#[serde(tag = "state", content = "id")]
|
pub struct Seed {
|
||||||
pub enum Seed {
|
pub nid: NodeId,
|
||||||
Disconnected(NodeId),
|
pub addrs: Vec<KnownAddress>,
|
||||||
Connected(NodeId),
|
pub state: Option<State>,
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Clone, Debug, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
|
impl Seed {
|
||||||
pub struct Seeds(BTreeSet<Seed>);
|
/// Check if this is a "connected" seed.
|
||||||
|
pub fn is_connected(&self) -> bool {
|
||||||
|
matches!(self.state, Some(State::Connected { .. }))
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn new(nid: NodeId, addrs: Vec<KnownAddress>, state: Option<State>) -> Self {
|
||||||
|
Self { nid, addrs, state }
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Clone, Debug, Default, serde::Serialize, serde::Deserialize)]
|
||||||
|
pub struct Seeds(BTreeMap<NodeId, Seed>);
|
||||||
|
|
||||||
impl Seeds {
|
impl Seeds {
|
||||||
pub fn insert(&mut self, seed: Seed) {
|
pub fn insert(&mut self, seed: Seed) {
|
||||||
self.0.insert(seed);
|
self.0.insert(seed.nid, seed);
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn connected(&self) -> impl Iterator<Item = &NodeId> {
|
/// Partitions the list of seeds into connected and disconnected seeds.
|
||||||
self.0.iter().filter_map(|s| match s {
|
/// Note that the disconnected seeds may be in a "connecting" state.
|
||||||
Seed::Connected(node) => Some(node),
|
pub fn partition(&self) -> (Vec<Seed>, Vec<Seed>) {
|
||||||
Seed::Disconnected(_) => None,
|
self.0.values().cloned().partition(|s| s.is_connected())
|
||||||
})
|
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn disconnected(&self) -> impl Iterator<Item = &NodeId> {
|
/// Return connected seeds.
|
||||||
self.0.iter().filter_map(|s| match s {
|
pub fn connected(&self) -> impl Iterator<Item = &Seed> {
|
||||||
Seed::Disconnected(node) => Some(node),
|
self.0.values().filter(|s| s.is_connected())
|
||||||
Seed::Connected(_) => None,
|
|
||||||
})
|
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn has_connections(&self) -> bool {
|
pub fn iter(&self) -> impl Iterator<Item = &Seed> {
|
||||||
self.0.iter().any(|s| match s {
|
self.0.values()
|
||||||
Seed::Connected(_) => true,
|
|
||||||
Seed::Disconnected(_) => false,
|
|
||||||
})
|
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn is_connected(&self, node: &NodeId) -> bool {
|
pub fn is_connected(&self, nid: &NodeId) -> bool {
|
||||||
self.0.contains(&Seed::Connected(*node))
|
self.0.get(nid).map_or(false, |s| s.is_connected())
|
||||||
}
|
|
||||||
|
|
||||||
pub fn is_disconnected(&self, node: &NodeId) -> bool {
|
|
||||||
self.0.contains(&Seed::Disconnected(*node))
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -90,7 +90,8 @@ pub struct Node {
|
||||||
}
|
}
|
||||||
|
|
||||||
/// A known address.
|
/// A known address.
|
||||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
|
||||||
|
#[serde(rename_all = "kebab-case")]
|
||||||
pub struct KnownAddress {
|
pub struct KnownAddress {
|
||||||
/// Network address.
|
/// Network address.
|
||||||
pub addr: Address,
|
pub addr: Address,
|
||||||
|
|
@ -115,7 +116,8 @@ impl KnownAddress {
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Address source. Specifies where an address originated from.
|
/// Address source. Specifies where an address originated from.
|
||||||
#[derive(Debug, Copy, Clone, PartialEq, Eq)]
|
#[derive(Debug, Copy, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
|
||||||
|
#[serde(rename_all = "kebab-case")]
|
||||||
pub enum Source {
|
pub enum Source {
|
||||||
/// An address that was shared by another peer.
|
/// An address that was shared by another peer.
|
||||||
Peer,
|
Peer,
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue