cli: Auto-connect on sync

This adds functionality so that on `rad sync` and `rad clone`, we
connect to known seeds if necessary, before attempting to fetch.

In passing, some related changes were made:

* The `rad sync` command's arguments were reworked:
  `--replicas <count>` can be used to specify a replica count when
    fetching
* The `rad patch` command no longer has a `--fetch` option
* It's now possible to tell the node not to automatically connect to
  peers in the background
* `AddressBook` is being used underneath `Seeds` to allow for shuffled
  iteration
This commit is contained in:
cloudhead 2023-08-09 10:48:58 +02:00
parent efb8de2728
commit d973fd42f2
No known key found for this signature in database
14 changed files with 429 additions and 149 deletions

View File

@ -0,0 +1,16 @@
If we're not connecting to seed nodes when cloning, the `clone` command will
automatically connect to the necessary seeds.
```
$ rad clone rad:z42hL2jL4XNk6K8oHQaSWfMgCL7ji
✓ Tracking relationship established for rad:z42hL2jL4XNk6K8oHQaSWfMgCL7ji with scope 'all'
✓ Connecting to z6MknSLrJoTcukLrE435hVNQT4JUhbvWLX4kUzqkEStBU8Vi[..]
✓ Fetching rad:z42hL2jL4XNk6K8oHQaSWfMgCL7ji from z6MknSL…StBU8Vi..
✓ Connecting to z6Mkt67GdsW7715MEfRuP4pSZxJRJh6kj6Y48WRqVv4N1tRk[..]
✓ Fetching rad:z42hL2jL4XNk6K8oHQaSWfMgCL7ji from z6Mkt67…v4N1tRk..
✓ Forking under z6Mkux1…nVhib7Z..
✓ Creating checkout in ./heartwood..
✓ Remote alice@z6MknSLrJoTcukLrE435hVNQT4JUhbvWLX4kUzqkEStBU8Vi added
✓ Remote-tracking branch alice@z6MknSLrJoTcukLrE435hVNQT4JUhbvWLX4kUzqkEStBU8Vi/master created for z6MknSL…StBU8Vi
✓ Repository successfully cloned under [..]/heartwood/
```

View File

@ -25,3 +25,41 @@ $ rad sync --announce --timeout 1
! Seed z6Mkux1aUQD2voWWukVb5nNUR7thrHveQG4pDQua8nVhib7Z timed out.. ! Seed z6Mkux1aUQD2voWWukVb5nNUR7thrHveQG4pDQua8nVhib7Z timed out..
✗ Sync failed: all seeds timed out ✗ Sync failed: all seeds timed out
``` ```
We can also use the `--fetch` option to only fetch objects:
```
$ rad sync --fetch
✓ Fetching rad:z42hL2jL4XNk6K8oHQaSWfMgCL7ji from z6Mkux1…nVhib7Z..
✓ Fetching rad:z42hL2jL4XNk6K8oHQaSWfMgCL7ji from z6Mkt67…v4N1tRk..
✓ Fetched repository from 2 seed(s)
```
Specifying both `--fetch` and `--announce` is equivalent to specifying none:
``` (fail)
$ rad sync --fetch --announce
✓ Fetching rad:z42hL2jL4XNk6K8oHQaSWfMgCL7ji from z6Mkux1…nVhib7Z..
✓ Fetching rad:z42hL2jL4XNk6K8oHQaSWfMgCL7ji from z6Mkt67…v4N1tRk..
✓ Fetched repository from 2 seed(s)
✗ Syncing with 2 node(s)..
! Seed z6Mkt67GdsW7715MEfRuP4pSZxJRJh6kj6Y48WRqVv4N1tRk timed out..
! Seed z6Mkux1aUQD2voWWukVb5nNUR7thrHveQG4pDQua8nVhib7Z timed out..
✗ Sync failed: all seeds timed out
```
It's also possible to use the `--seed` flag to only sync with a specific node:
```
$ rad sync --fetch --seed z6Mkt67GdsW7715MEfRuP4pSZxJRJh6kj6Y48WRqVv4N1tRk
✓ Fetching rad:z42hL2jL4XNk6K8oHQaSWfMgCL7ji from z6Mkt67…v4N1tRk..
✓ Fetched repository from 1 seed(s)
```
And the `--replicas` flag to sync with a number of nodes:
```
$ rad sync --fetch --replicas 1
✓ Fetching rad:z42hL2jL4XNk6K8oHQaSWfMgCL7ji from z6Mkux1…nVhib7Z..
✓ Fetched repository from 1 seed(s)
```

View File

@ -2,6 +2,7 @@
use std::ffi::OsString; use std::ffi::OsString;
use std::path::Path; use std::path::Path;
use std::str::FromStr; use std::str::FromStr;
use std::time;
use anyhow::anyhow; use anyhow::anyhow;
use thiserror::Error; use thiserror::Error;
@ -183,7 +184,12 @@ pub fn clone<G: Signer>(
); );
} }
let results = sync::fetch_all(id, node)?; let results = sync::fetch(
id,
sync::SyncMode::default(),
time::Duration::from_secs(9),
node,
)?;
let Ok(repository) = storage.repository(id) else { let Ok(repository) = storage.repository(id) else {
// If we don't have the project locally, even after attempting to fetch, // If we don't have the project locally, even after attempting to fetch,
// there's nothing we can do. // there's nothing we can do.

View File

@ -25,10 +25,9 @@ use anyhow::anyhow;
use radicle::cob::patch; use radicle::cob::patch;
use radicle::cob::patch::PatchId; use radicle::cob::patch::PatchId;
use radicle::prelude::*;
use radicle::storage::git::transport; use radicle::storage::git::transport;
use radicle::{prelude::*, Node};
use crate::commands::rad_sync as sync;
use crate::git::Rev; use crate::git::Rev;
use crate::terminal as term; use crate::terminal as term;
use crate::terminal::args::{string, Args, Error, Help}; use crate::terminal::args::{string, Args, Error, Help};
@ -158,7 +157,6 @@ pub enum Operation {
#[derive(Debug)] #[derive(Debug)]
pub struct Options { pub struct Options {
pub op: Operation, pub op: Operation,
pub fetch: bool,
pub announce: bool, pub announce: bool,
pub push: bool, pub push: bool,
pub verbose: bool, pub verbose: bool,
@ -171,7 +169,6 @@ impl Args for Options {
let mut parser = lexopt::Parser::from_args(args); let mut parser = lexopt::Parser::from_args(args);
let mut op: Option<OperationName> = None; let mut op: Option<OperationName> = None;
let mut verbose = false; let mut verbose = false;
let mut fetch = false;
let mut announce = false; let mut announce = false;
let mut patch_id = None; let mut patch_id = None;
let mut revision_id = None; let mut revision_id = None;
@ -194,12 +191,6 @@ impl Args for Options {
Long("no-message") => { Long("no-message") => {
message = Message::Blank; message = Message::Blank;
} }
Long("fetch") => {
fetch = true;
}
Long("no-fetch") => {
fetch = false;
}
Long("announce") => { Long("announce") => {
announce = true; announce = true;
} }
@ -327,7 +318,6 @@ impl Args for Options {
Ok(( Ok((
Options { Options {
op, op,
fetch,
push, push,
verbose, verbose,
announce, announce,
@ -346,10 +336,6 @@ pub fn run(options: Options, ctx: impl term::Context) -> anyhow::Result<()> {
transport::local::register(profile.storage.clone()); transport::local::register(profile.storage.clone());
if options.fetch {
sync::fetch_all(repository.id(), &mut Node::new(profile.socket()))?;
}
match options.op { match options.op {
Operation::List { filter: Filter(f) } => { Operation::List { filter: Filter(f) } => {
list::run(f, &repository, &profile)?; list::run(f, &repository, &profile)?;

View File

@ -6,7 +6,7 @@ use anyhow::{anyhow, Context as _};
use radicle::node; use radicle::node;
use radicle::node::{FetchResult, FetchResults, Handle as _, Node}; use radicle::node::{FetchResult, FetchResults, Handle as _, Node};
use radicle::prelude::{Id, NodeId, Profile}; use radicle::prelude::{Id, NodeId};
use crate::terminal as term; use crate::terminal as term;
use crate::terminal::args::{Args, Error, Help}; use crate::terminal::args::{Args, Error, Help};
@ -19,35 +19,57 @@ pub const HELP: Help = Help {
Usage Usage
rad sync [<rid>] [<option>...] rad sync [<rid>] [<option>...]
rad sync [<rid>] [--fetch] [--seed <nid>] [<option>...] rad sync [<rid>] [--fetch] [<rid>] [<option>...]
rad sync [<rid>] [--announce] [<option>...] rad sync [<rid>] [--announce] [<rid>] [<option>...]
By default, the current repository is synchronized both ways. By default, the current repository is synchronized both ways.
If an <rid> is specified, that repository is synced instead.
The process begins by fetching changes from connected seeds, The process begins by fetching changes from connected seeds,
followed by announcing local refs to peers, thereby prompting followed by announcing local refs to peers, thereby prompting
them to fetch from us. them to fetch from us.
When `--fetch` is specified, a seed may be given with the `--seed` When `--fetch` is specified, any number of seeds may be given
option. using the `--seed` option, eg. `--seed <nid>@<addr>:<port>`.
When either `--fetch` or `--announce` are specified, this command When `--replicas` is specified, the given replication factor will try
to be matched. For example, `--replicas 5` will sync with 5 seeds.
When `--fetch` or `--announce` are specified on their own, this command
will only fetch or announce. will only fetch or announce.
Options Options
--fetch, -f Fetch from seeds --fetch, -f Turn on fetching (default: true)
--announce, -a Announce refs to seeds --announce, -a Turn on announcing (default: true)
--seed <nid> Seed to fetch from (use with `--fetch`) --timeout <secs> How many seconds to wait while syncing
--timeout <secs> How many seconds to wait while syncing --seed <nid> Sync with the given node (may be specified multiple times)
--verbose, -v Verbose output --replicas, -r <count> Sync with a specific number of seeds
--help Print help --verbose, -v Verbose output
--help Print help
"#, "#,
}; };
#[derive(Default, Debug, PartialEq, Eq)] #[derive(Debug, Default, PartialEq, Eq)]
pub struct SyncOptions {
mode: SyncMode,
direction: SyncDirection,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum SyncMode { pub enum SyncMode {
Replicas(usize),
Seeds(Vec<NodeId>),
}
impl Default for SyncMode {
fn default() -> Self {
Self::Replicas(3)
}
}
#[derive(Debug, Default, PartialEq, Eq)]
pub enum SyncDirection {
Fetch, Fetch,
Announce, Announce,
#[default] #[default]
@ -57,10 +79,9 @@ pub enum SyncMode {
#[derive(Default, Debug)] #[derive(Default, Debug)]
pub struct Options { pub struct Options {
pub rid: Option<Id>, pub rid: Option<Id>,
pub seed: Option<NodeId>,
pub verbose: bool, pub verbose: bool,
pub timeout: time::Duration, pub timeout: time::Duration,
pub mode: SyncMode, pub sync: SyncOptions,
} }
impl Args for Options { impl Args for Options {
@ -71,24 +92,46 @@ impl Args for Options {
let mut verbose = false; let mut verbose = false;
let mut timeout = time::Duration::from_secs(9); let mut timeout = time::Duration::from_secs(9);
let mut rid = None; let mut rid = None;
let mut seed = None; let mut sync = SyncOptions::default();
let mut mode = SyncMode::default();
while let Some(arg) = parser.next()? { while let Some(arg) = parser.next()? {
match arg { match arg {
Long("verbose") | Short('v') => { Long("verbose") | Short('v') => {
verbose = true; verbose = true;
} }
Long("fetch") | Short('f') => {
sync.direction = match sync.direction {
SyncDirection::Both => SyncDirection::Fetch,
SyncDirection::Announce => SyncDirection::Both,
SyncDirection::Fetch => SyncDirection::Fetch,
};
}
Long("replicas") | Short('r') => {
let val = parser.value()?;
let count = term::args::number(&val)?;
if let SyncMode::Replicas(ref mut r) = sync.mode {
*r = count;
} else {
anyhow::bail!("`--replicas` (-r) cannot be specified with `--seed`");
}
}
Long("seed") => { Long("seed") => {
let val = parser.value()?; let val = parser.value()?;
let val = term::args::nid(&val)?; let nid = term::args::nid(&val)?;
seed = Some(val);
if let SyncMode::Seeds(ref mut seeds) = sync.mode {
seeds.push(nid);
} else {
sync.mode = SyncMode::Seeds(vec![nid]);
}
} }
Long("fetch") | Short('f') if mode == SyncMode::Both => { Long("announce") | Short('a') => {
mode = SyncMode::Fetch; sync.direction = match sync.direction {
} SyncDirection::Both => SyncDirection::Announce,
Long("announce") | Short('a') if mode == SyncMode::Both => { SyncDirection::Announce => SyncDirection::Announce,
mode = SyncMode::Announce; SyncDirection::Fetch => SyncDirection::Both,
};
} }
Long("timeout") | Short('t') => { Long("timeout") | Short('t') => {
let value = parser.value()?; let value = parser.value()?;
@ -108,8 +151,10 @@ impl Args for Options {
} }
} }
if seed.is_some() && mode != SyncMode::Fetch { if sync.direction == SyncDirection::Announce {
anyhow::bail!("`--seed` must be used with `--fetch`"); if let SyncMode::Seeds(_) = sync.mode {
anyhow::bail!("`--seed` is only supported when fetching.");
}
} }
Ok(( Ok((
@ -117,8 +162,7 @@ impl Args for Options {
rid, rid,
verbose, verbose,
timeout, timeout,
seed, sync,
mode,
}, },
vec![], vec![],
)) ))
@ -136,22 +180,35 @@ pub fn run(options: Options, ctx: impl term::Context) -> anyhow::Result<()> {
rid rid
} }
}; };
let mut node = radicle::Node::new(profile.socket()); let mut node = radicle::Node::new(profile.socket());
let mode = options.sync.mode;
match options.mode { if [SyncDirection::Fetch, SyncDirection::Both].contains(&options.sync.direction) {
SyncMode::Announce => announce(rid, options.timeout, node), if !profile.tracking()?.is_repo_tracked(&rid)? {
SyncMode::Fetch => fetch(rid, options.seed, &mut node, profile), anyhow::bail!("repository {rid} is not tracked");
SyncMode::Both => { }
fetch(rid, options.seed, &mut node, profile)?; let results = fetch(rid, mode.clone(), options.timeout, &mut node)?;
announce(rid, options.timeout, node)?; let success = results.success().count();
let failed = results.failed().count();
Ok(()) if success == 0 {
term::error(format!("Failed to fetch repository from {failed} seed(s)"));
} else {
term::success!("Fetched repository from {success} seed(s)");
} }
} }
if [SyncDirection::Announce, SyncDirection::Both].contains(&options.sync.direction) {
announce(rid, mode, options.timeout, node)?;
}
Ok(())
} }
fn announce(rid: Id, timeout: time::Duration, mut node: Node) -> anyhow::Result<()> { fn announce(
rid: Id,
_mode: SyncMode,
timeout: time::Duration,
mut node: Node,
) -> anyhow::Result<()> {
let seeds = node.seeds(rid)?; let seeds = node.seeds(rid)?;
let connected = seeds.connected().map(|s| s.nid).collect::<Vec<_>>(); let connected = seeds.connected().map(|s| s.nid).collect::<Vec<_>>();
if connected.is_empty() { if connected.is_empty() {
@ -184,45 +241,80 @@ fn announce(rid: Id, timeout: time::Duration, mut node: Node) -> anyhow::Result<
pub fn fetch( pub fn fetch(
rid: Id, rid: Id,
seed: Option<NodeId>, mode: SyncMode,
timeout: time::Duration,
node: &mut Node, node: &mut Node,
profile: Profile, ) -> Result<FetchResults, node::Error> {
) -> anyhow::Result<()> { match mode {
if !profile.tracking()?.is_repo_tracked(&rid)? { SyncMode::Seeds(seeds) => {
anyhow::bail!("repository {rid} is not tracked"); let mut results = FetchResults::default();
for seed in seeds {
let result = fetch_from(rid, &seed, node)?;
results.push(seed, result);
}
Ok(results)
}
SyncMode::Replicas(count) => fetch_all(rid, count, timeout, node),
} }
let results = if let Some(seed) = seed {
let result = fetch_from(rid, &seed, node)?;
FetchResults::from(vec![(seed, result)])
} else {
fetch_all(rid, node)?
};
let success = results.success().count();
let failed = results.failed().count();
if success == 0 {
term::error(format!("Failed to fetch repository from {failed} seed(s)"));
} else {
term::success!("Fetched repository from {success} seed(s)");
}
Ok(())
} }
pub fn fetch_all(rid: Id, node: &mut Node) -> Result<FetchResults, node::Error> { fn fetch_all(
rid: Id,
count: usize,
timeout: time::Duration,
node: &mut Node,
) -> Result<FetchResults, node::Error> {
// Get seeds. This consults the local routing table only. // Get seeds. This consults the local routing table only.
let seeds = node.seeds(rid)?; let seeds = node.seeds(rid)?;
let mut results = FetchResults::default(); let mut results = FetchResults::default();
let (connected, mut disconnected) = seeds.partition();
// Fetch from connected seeds. // Fetch from connected seeds.
for seed in seeds.connected() { for seed in connected.iter().take(count) {
let result = fetch_from(rid, &seed.nid, node)?; let result = fetch_from(rid, &seed.nid, node)?;
results.push(seed.nid, result); results.push(seed.nid, result);
} }
// Try to connect to disconnected seeds and fetch from them.
while results.success().count() < count {
let Some(seed) = disconnected.pop() else {
break;
};
// Try all seed addresses until one succeeds.
for ka in seed.addrs {
let spinner = term::spinner(format!(
"Connecting to {}@{}..",
term::format::tertiary(&seed.nid),
term::format::tertiary(&ka.addr)
));
let cr = node.connect(
seed.nid,
ka.addr,
node::ConnectOptions {
persistent: false,
timeout,
},
)?;
match cr {
node::ConnectResult::Connected => {
spinner.finish();
let result = fetch_from(rid, &seed.nid, node)?;
results.push(seed.nid, result);
break;
}
node::ConnectResult::Disconnected { .. } => {
spinner.failed();
continue;
}
}
}
}
Ok(results) Ok(results)
} }
pub fn fetch_from(rid: Id, seed: &NodeId, node: &mut Node) -> Result<FetchResult, node::Error> { fn fetch_from(rid: Id, seed: &NodeId, node: &mut Node) -> Result<FetchResult, node::Error> {
let spinner = term::spinner(format!( let spinner = term::spinner(format!(
"Fetching {} from {}..", "Fetching {} from {}..",
term::format::tertiary(rid), term::format::tertiary(rid),

View File

@ -1,4 +1,5 @@
use std::ffi::OsString; use std::ffi::OsString;
use std::time;
use anyhow::anyhow; use anyhow::anyhow;
@ -123,7 +124,12 @@ pub fn run(options: Options, ctx: impl term::Context) -> anyhow::Result<()> {
track_repo(rid, scope, &mut node)?; track_repo(rid, scope, &mut node)?;
if options.fetch { if options.fetch {
sync::fetch(rid, None, &mut node, profile)?; sync::fetch(
rid,
sync::SyncMode::default(),
time::Duration::from_secs(6),
&mut node,
)?;
} }
} }
} }

View File

@ -3,6 +3,9 @@ use std::str::FromStr;
use std::{env, thread, time}; use std::{env, thread, time};
use radicle::git; use radicle::git;
use radicle::node;
use radicle::node::address::Store as _;
use radicle::node::routing::Store as _;
use radicle::node::Alias; use radicle::node::Alias;
use radicle::node::Handle as _; use radicle::node::Handle as _;
use radicle::prelude::Id; use radicle::prelude::Id;
@ -512,6 +515,77 @@ fn rad_clone_all() {
eve.has_inventory_of(&acme, &bob.id); eve.has_inventory_of(&acme, &bob.id);
} }
#[test]
fn rad_clone_connect() {
let mut environment = Environment::new();
let working = environment.tmp().join("working");
let alice = environment.node(Config::test(Alias::new("alice")));
let bob = environment.node(Config::test(Alias::new("bob")));
let mut eve = environment.node(Config::test(Alias::new("eve")));
let acme = Id::from_str("z42hL2jL4XNk6K8oHQaSWfMgCL7ji").unwrap();
let now = localtime::LocalTime::now().as_secs();
fixtures::repository(working.join("acme"));
test(
"examples/rad-init.md",
working.join("acme"),
Some(&alice.home),
[],
)
.unwrap();
let mut alice = alice.spawn();
let mut bob = bob.spawn();
// Let Eve know about Alice and Bob having the repo.
eve.routing.insert([&acme], alice.id, now).unwrap();
eve.routing.insert([&acme], bob.id, now).unwrap();
eve.addresses
.insert(
&alice.id,
node::Features::SEED,
Alias::new("alice"),
0,
now,
[node::KnownAddress::new(
node::Address::from(alice.addr),
node::address::Source::Imported,
)],
)
.unwrap();
eve.addresses
.insert(
&bob.id,
node::Features::SEED,
Alias::new("bob"),
0,
now,
[node::KnownAddress::new(
node::Address::from(bob.addr),
node::address::Source::Imported,
)],
)
.unwrap();
eve.config.peers = node::config::PeerConfig::Static;
let eve = eve.spawn();
bob.handle.track_repo(acme, Scope::All).unwrap();
alice.connect(&bob);
bob.routes_to(&[(acme, alice.id)]);
eve.routes_to(&[(acme, alice.id), (acme, bob.id)]);
alice.routes_to(&[(acme, alice.id), (acme, bob.id)]);
test(
"examples/rad-clone-connect.md",
working.join("acme"),
Some(&eve.home),
[],
)
.unwrap();
}
#[test] #[test]
fn rad_self() { fn rad_self() {
let mut environment = Environment::new(); let mut environment = Environment::new();

View File

@ -152,7 +152,7 @@ impl TryFrom<String> for Signature {
} }
/// The public/verification key. /// The public/verification key.
#[derive(Serialize, Deserialize, Eq, Copy, Clone)] #[derive(Hash, Serialize, Deserialize, PartialEq, Eq, Copy, Clone)]
#[serde(into = "String", try_from = "String")] #[serde(into = "String", try_from = "String")]
pub struct PublicKey(pub ed25519::PublicKey); pub struct PublicKey(pub ed25519::PublicKey);
@ -282,12 +282,6 @@ pub enum PublicKeyError {
InvalidKey(#[from] ed25519::Error), InvalidKey(#[from] ed25519::Error),
} }
impl std::hash::Hash for PublicKey {
fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
self.0.deref().hash(state)
}
}
impl PartialOrd for PublicKey { impl PartialOrd for PublicKey {
fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> { fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
self.0.as_ref().partial_cmp(other.as_ref()) self.0.as_ref().partial_cmp(other.as_ref())
@ -318,12 +312,6 @@ impl fmt::Debug for PublicKey {
} }
} }
impl PartialEq for PublicKey {
fn eq(&self, other: &Self) -> bool {
self.0 == other.0
}
}
impl From<ed25519::PublicKey> for PublicKey { impl From<ed25519::PublicKey> for PublicKey {
fn from(other: ed25519::PublicKey) -> Self { fn from(other: ed25519::PublicKey) -> Self {
Self(other) Self(other)

View File

@ -22,6 +22,7 @@ use nonempty::NonEmpty;
use radicle::node::address; use radicle::node::address;
use radicle::node::address::{AddressBook, KnownAddress}; use radicle::node::address::{AddressBook, KnownAddress};
use radicle::node::config::PeerConfig;
use radicle::node::ConnectOptions; use radicle::node::ConnectOptions;
use crate::crypto; use crate::crypto;
@ -53,8 +54,6 @@ use self::limitter::RateLimiter;
use self::message::InventoryAnnouncement; use self::message::InventoryAnnouncement;
use self::tracking::NamespacesError; use self::tracking::NamespacesError;
/// Target number of peers to maintain connections to.
pub const TARGET_OUTBOUND_PEERS: usize = 8;
/// How often to run the "idle" task. /// How often to run the "idle" task.
pub const IDLE_INTERVAL: LocalDuration = LocalDuration::from_secs(30); pub const IDLE_INTERVAL: LocalDuration = LocalDuration::from_secs(30);
/// How often to run the "announce" task. /// How often to run the "announce" task.
@ -1359,30 +1358,31 @@ where
} }
fn seeds(&self, rid: &Id) -> Result<Seeds, Error> { fn seeds(&self, rid: &Id) -> Result<Seeds, Error> {
let seeds = match self.routing.get(rid) { match self.routing.get(rid) {
Ok(seeds) => seeds.into_iter().fold(Seeds::default(), |mut seeds, node| { Ok(seeds) => {
if node != self.node_id() { Ok(seeds
let addrs: Vec<KnownAddress> = self .into_iter()
.addresses .fold(Seeds::new(self.rng.clone()), |mut seeds, node| {
.get(&node) if node != self.node_id() {
.ok() let addrs: Vec<KnownAddress> = self
.flatten() .addresses
.map(|n| n.addrs) .get(&node)
.unwrap_or(vec![]); .ok()
.flatten()
.map(|n| n.addrs)
.unwrap_or(vec![]);
if let Some(s) = self.sessions.get(&node) { if let Some(s) = self.sessions.get(&node) {
seeds.insert(Seed::new(node, addrs, Some(s.state.clone()))); seeds.insert(Seed::new(node, addrs, Some(s.state.clone())));
} else { } else {
seeds.insert(Seed::new(node, addrs, None)); seeds.insert(Seed::new(node, addrs, None));
} }
} }
seeds seeds
}), }))
Err(err) => {
return Err(Error::Routing(err));
} }
}; Err(err) => Err(Error::Routing(err)),
Ok(seeds) }
} }
/// Return a new filter object, based on our tracking policy. /// Return a new filter object, based on our tracking policy.
@ -1513,6 +1513,9 @@ where
} }
fn maintain_connections(&mut self) { fn maintain_connections(&mut self) {
let PeerConfig::Dynamic { target } = self.config.peers else {
return;
};
trace!(target: "service", "Maintaining connections.."); trace!(target: "service", "Maintaining connections..");
let now = self.clock; let now = self.clock;
@ -1522,7 +1525,7 @@ where
.filter(|s| s.link.is_outbound()) .filter(|s| s.link.is_outbound())
.filter(|s| s.is_connected() || s.is_connecting()) .filter(|s| s.is_connected() || s.is_connecting())
.count(); .count();
let wanted = TARGET_OUTBOUND_PEERS.saturating_sub(outbound); let wanted = target.saturating_sub(outbound);
// Don't connect to more peers than needed. // Don't connect to more peers than needed.
if wanted == 0 { if wanted == 0 {

View File

@ -19,9 +19,10 @@ use radicle::git;
use radicle::git::refname; use radicle::git::refname;
use radicle::identity::Id; use radicle::identity::Id;
use radicle::node::address::Book; use radicle::node::address::Book;
use radicle::node::routing;
use radicle::node::routing::Store; use radicle::node::routing::Store;
use radicle::node::tracking::store as TrackingStore; use radicle::node::tracking::store as tracking;
use radicle::node::{Alias, ADDRESS_DB_FILE, TRACKING_DB_FILE}; use radicle::node::{Alias, ADDRESS_DB_FILE, ROUTING_DB_FILE, TRACKING_DB_FILE};
use radicle::node::{ConnectOptions, Handle as _}; use radicle::node::{ConnectOptions, Handle as _};
use radicle::profile; use radicle::profile;
use radicle::profile::Home; use radicle::profile::Home;
@ -79,16 +80,24 @@ impl Environment {
pub fn node(&mut self, config: Config) -> Node<MemorySigner> { pub fn node(&mut self, config: Config) -> Node<MemorySigner> {
let profile = self.profile(&config.alias); let profile = self.profile(&config.alias);
let signer = MemorySigner::load(&profile.keystore, None).unwrap(); let signer = MemorySigner::load(&profile.keystore, None).unwrap();
let tracking_db = profile.home.node().join(TRACKING_DB_FILE); let tracking_db = profile.home.node().join(TRACKING_DB_FILE);
TrackingStore::Config::open(tracking_db).unwrap(); let tracking = tracking::Config::open(tracking_db).unwrap();
let routing_db = profile.home.node().join(ROUTING_DB_FILE);
let routing = routing::Table::open(routing_db).unwrap();
let addresses_db = profile.home.node().join(ADDRESS_DB_FILE); let addresses_db = profile.home.node().join(ADDRESS_DB_FILE);
Book::open(addresses_db).unwrap(); let addresses = Book::open(addresses_db).unwrap();
Node { Node {
id: *profile.id(), id: *profile.id(),
home: profile.home, home: profile.home,
config, config,
signer, signer,
addresses,
routing,
tracking,
storage: profile.storage, storage: profile.storage,
} }
} }
@ -104,7 +113,7 @@ impl Environment {
let alias = Alias::from_str(alias).unwrap(); let alias = Alias::from_str(alias).unwrap();
let config = profile::Config::init(alias, &home.config()).unwrap(); let config = profile::Config::init(alias, &home.config()).unwrap();
TrackingStore::Config::open(tracking_db).unwrap(); tracking::Config::open(tracking_db).unwrap();
let addresses_db = home.node().join(ADDRESS_DB_FILE); let addresses_db = home.node().join(ADDRESS_DB_FILE);
Book::open(addresses_db).unwrap(); Book::open(addresses_db).unwrap();
@ -131,6 +140,9 @@ pub struct Node<G> {
pub signer: G, pub signer: G,
pub storage: Storage, pub storage: Storage,
pub config: Config, pub config: Config,
pub addresses: Book,
pub routing: routing::Table,
pub tracking: tracking::Config<tracking::Write>,
} }
/// Handle to a running node. /// Handle to a running node.
@ -320,6 +332,9 @@ impl Node<MockSigner> {
let home = Home::new(home).unwrap(); let home = Home::new(home).unwrap();
let signer = MockSigner::default(); let signer = MockSigner::default();
let storage = Storage::open(home.storage()).unwrap(); let storage = Storage::open(home.storage()).unwrap();
let addresses = Book::memory().unwrap();
let tracking = tracking::Config::<tracking::Write>::memory().unwrap();
let routing = routing::Table::memory().unwrap();
Self { Self {
id: *signer.public_key(), id: *signer.public_key(),
@ -327,6 +342,9 @@ impl Node<MockSigner> {
signer, signer,
storage, storage,
config, config,
addresses,
tracking,
routing,
} }
} }
} }

View File

@ -6,7 +6,7 @@ pub mod events;
pub mod routing; pub mod routing;
pub mod tracking; pub mod tracking;
use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet}; use std::collections::{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;
@ -23,6 +23,7 @@ use serde_json as json;
use crate::crypto::PublicKey; use crate::crypto::PublicKey;
use crate::identity::Id; use crate::identity::Id;
use crate::profile;
use crate::storage::RefUpdate; use crate::storage::RefUpdate;
pub use address::KnownAddress; pub use address::KnownAddress;
@ -66,7 +67,7 @@ pub enum PingState {
Ok, Ok,
} }
#[derive(Debug, Clone, Serialize, Deserialize)] #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[allow(clippy::large_enum_variant)] #[allow(clippy::large_enum_variant)]
pub enum State { pub enum State {
/// Initial state for outgoing connections. /// Initial state for outgoing connections.
@ -401,7 +402,7 @@ pub struct Session {
pub state: State, pub state: State,
} }
#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)] #[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")] #[serde(rename_all = "camelCase")]
pub struct Seed { pub struct Seed {
pub nid: NodeId, pub nid: NodeId,
@ -420,10 +421,18 @@ impl Seed {
} }
} }
#[derive(Clone, Debug, Default, serde::Serialize, serde::Deserialize)] #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
pub struct Seeds(BTreeMap<NodeId, Seed>); /// Represents a set of seeds with associated metadata. Uses an RNG
/// underneath, so every iteration returns a different ordering.
pub struct Seeds(address::AddressBook<NodeId, Seed>);
impl Seeds { impl Seeds {
/// Create a new seeds list from an RNG.
pub fn new(rng: fastrand::Rng) -> Self {
Self(address::AddressBook::new(rng))
}
/// Insert a seed.
pub fn insert(&mut self, seed: Seed) { pub fn insert(&mut self, seed: Seed) {
self.0.insert(seed.nid, seed); self.0.insert(seed.nid, seed);
} }
@ -431,21 +440,30 @@ impl Seeds {
/// Partitions the list of seeds into connected and disconnected seeds. /// Partitions the list of seeds into connected and disconnected seeds.
/// Note that the disconnected seeds may be in a "connecting" state. /// Note that the disconnected seeds may be in a "connecting" state.
pub fn partition(&self) -> (Vec<Seed>, Vec<Seed>) { pub fn partition(&self) -> (Vec<Seed>, Vec<Seed>) {
self.0.values().cloned().partition(|s| s.is_connected()) self.0
.shuffled()
.map(|(_, v)| v)
.cloned()
.partition(|s| s.is_connected())
} }
/// Return connected seeds. /// Return connected seeds.
pub fn connected(&self) -> impl Iterator<Item = &Seed> { pub fn connected(&self) -> impl Iterator<Item = &Seed> {
self.0.values().filter(|s| s.is_connected()) self.0
} .shuffled()
.map(|(_, v)| v)
pub fn iter(&self) -> impl Iterator<Item = &Seed> { .filter(|s| s.is_connected())
self.0.values()
} }
/// Check if a seed is connected.
pub fn is_connected(&self, nid: &NodeId) -> bool { pub fn is_connected(&self, nid: &NodeId) -> bool {
self.0.get(nid).map_or(false, |s| s.is_connected()) self.0.get(nid).map_or(false, |s| s.is_connected())
} }
/// Return a new seeds object with the given RNG.
pub fn with(self, rng: fastrand::Rng) -> Self {
Self(self.0.with(rng))
}
} }
/// Announcement result returned by [`Node::announce`]. /// Announcement result returned by [`Node::announce`].
@ -803,7 +821,7 @@ impl Handle for Node {
.next() .next()
.ok_or(Error::EmptyResponse)??; .ok_or(Error::EmptyResponse)??;
Ok(seeds) Ok(seeds.with(profile::env::rng()))
} }
fn fetch(&mut self, rid: Id, from: NodeId) -> Result<FetchResult, Error> { fn fetch(&mut self, rid: Id, from: NodeId) -> Result<FetchResult, Error> {

View File

@ -1,3 +1,4 @@
use std::hash;
use std::ops::{Deref, DerefMut}; use std::ops::{Deref, DerefMut};
use localtime::LocalTime; use localtime::LocalTime;
@ -9,13 +10,15 @@ use crate::node::{Address, Alias};
use crate::prelude::Timestamp; use crate::prelude::Timestamp;
/// A map with the ability to randomly select values. /// A map with the ability to randomly select values.
#[derive(Debug, Clone)] #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct AddressBook<K, V> { #[serde(transparent)]
pub struct AddressBook<K: hash::Hash + Eq, V> {
inner: RandomMap<K, V>, inner: RandomMap<K, V>,
#[serde(skip)]
rng: fastrand::Rng, rng: fastrand::Rng,
} }
impl<K, V> AddressBook<K, V> { impl<K: hash::Hash + Eq, V> AddressBook<K, V> {
/// Create a new address book. /// Create a new address book.
pub fn new(rng: fastrand::Rng) -> Self { pub fn new(rng: fastrand::Rng) -> Self {
Self { Self {
@ -53,14 +56,22 @@ impl<K, V> AddressBook<K, V> {
/// Return a shuffled iterator over the keys. /// Return a shuffled iterator over the keys.
pub fn shuffled(&self) -> std::vec::IntoIter<(&K, &V)> { pub fn shuffled(&self) -> std::vec::IntoIter<(&K, &V)> {
let mut keys = self.inner.iter().collect::<Vec<_>>(); let mut items = self.inner.iter().collect::<Vec<_>>();
self.rng.shuffle(&mut keys); self.rng.shuffle(&mut items);
keys.into_iter() items.into_iter()
}
/// Return a new address book with the given RNG.
pub fn with(self, rng: fastrand::Rng) -> Self {
Self {
inner: self.inner,
rng,
}
} }
} }
impl<K, V> Deref for AddressBook<K, V> { impl<K: hash::Hash + Eq, V> Deref for AddressBook<K, V> {
type Target = RandomMap<K, V>; type Target = RandomMap<K, V>;
fn deref(&self) -> &Self::Target { fn deref(&self) -> &Self::Target {
@ -68,7 +79,7 @@ impl<K, V> Deref for AddressBook<K, V> {
} }
} }
impl<K, V> DerefMut for AddressBook<K, V> { impl<K: hash::Hash + Eq, V> DerefMut for AddressBook<K, V> {
fn deref_mut(&mut self) -> &mut Self::Target { fn deref_mut(&mut self) -> &mut Self::Target {
&mut self.inner &mut self.inner
} }

View File

@ -8,6 +8,9 @@ use crate::node;
use crate::node::tracking::{Policy, Scope}; use crate::node::tracking::{Policy, Scope};
use crate::node::{Address, Alias, NodeId}; use crate::node::{Address, Alias, NodeId};
/// Target number of peers to maintain connections to.
pub const TARGET_OUTBOUND_PEERS: usize = 8;
/// Peer-to-peer network. /// Peer-to-peer network.
#[derive(Default, Debug, Copy, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] #[derive(Default, Debug, Copy, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")] #[serde(rename_all = "camelCase")]
@ -103,12 +106,32 @@ impl Deref for ConnectAddress {
} }
} }
/// Peer configuration.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase", tag = "type")]
pub enum PeerConfig {
/// Static peer set. Connect to the configured peers and maintain the connections.
Static,
/// Dynamic peer set.
Dynamic { target: usize },
}
impl Default for PeerConfig {
fn default() -> Self {
Self::Dynamic {
target: TARGET_OUTBOUND_PEERS,
}
}
}
/// Service configuration. /// Service configuration.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")] #[serde(rename_all = "camelCase")]
pub struct Config { pub struct Config {
/// Node alias. /// Node alias.
pub alias: Alias, pub alias: Alias,
/// Peer configuration.
pub peers: PeerConfig,
/// Peers to connect to on startup. /// Peers to connect to on startup.
/// Connections to these peers will be maintained. /// Connections to these peers will be maintained.
pub connect: HashSet<ConnectAddress>, pub connect: HashSet<ConnectAddress>,
@ -137,6 +160,7 @@ impl Config {
pub fn new(alias: Alias) -> Self { pub fn new(alias: Alias) -> Self {
Self { Self {
alias, alias,
peers: PeerConfig::default(),
connect: HashSet::default(), connect: HashSet::default(),
external_addresses: vec![], external_addresses: vec![],
network: Network::default(), network: Network::default(),

View File

@ -50,13 +50,13 @@ pub mod env {
/// Get a random number generator from the environment. /// Get a random number generator from the environment.
pub fn rng() -> fastrand::Rng { pub fn rng() -> fastrand::Rng {
let Ok(seed) = std::env::var(RAD_RNG_SEED) else { if let Ok(seed) = std::env::var(RAD_RNG_SEED) {
return fastrand::Rng::new(); return fastrand::Rng::with_seed(
}; seed.parse()
fastrand::Rng::with_seed( .expect("env::rng: invalid seed specified in `RAD_RNG_SEED`"),
seed.parse() );
.expect("env::rng: invalid seed specified in `RAD_RNG_SEED`"), }
) fastrand::Rng::new()
} }
} }