268 lines
8.0 KiB
Rust
268 lines
8.0 KiB
Rust
use core::time;
|
|
use std::collections::BTreeSet;
|
|
use std::io;
|
|
use std::io::Write;
|
|
|
|
use radicle::node::sync;
|
|
use radicle::node::{Handle as _, NodeId};
|
|
use radicle::storage::{ReadRepository, RepositoryError};
|
|
use radicle::{Node, Profile};
|
|
|
|
use crate::terminal as term;
|
|
|
|
/// Default time to wait for syncing to complete.
|
|
pub const DEFAULT_SYNC_TIMEOUT: time::Duration = time::Duration::from_secs(9);
|
|
|
|
/// Repository sync settings.
|
|
#[derive(Debug, Clone, PartialEq, Eq)]
|
|
pub struct SyncSettings {
|
|
/// Sync with at least N replicas.
|
|
pub replicas: sync::ReplicationFactor,
|
|
/// Sync with the given list of seeds.
|
|
pub seeds: BTreeSet<NodeId>,
|
|
/// How long to wait for syncing to complete.
|
|
pub timeout: time::Duration,
|
|
}
|
|
|
|
impl SyncSettings {
|
|
/// Set sync timeout. Defaults to [`DEFAULT_SYNC_TIMEOUT`].
|
|
pub fn timeout(mut self, timeout: time::Duration) -> Self {
|
|
self.timeout = timeout;
|
|
self
|
|
}
|
|
|
|
/// Set replicas.
|
|
pub fn replicas(mut self, replicas: sync::ReplicationFactor) -> Self {
|
|
self.replicas = replicas;
|
|
self
|
|
}
|
|
|
|
/// Set seeds.
|
|
pub fn seeds(mut self, seeds: impl IntoIterator<Item = NodeId>) -> Self {
|
|
self.seeds = seeds.into_iter().collect();
|
|
self
|
|
}
|
|
|
|
/// Use profile to populate sync settings, by adding preferred seeds if no seeds are specified,
|
|
/// and removing the local node from the set.
|
|
pub fn with_profile(mut self, profile: &Profile) -> Self {
|
|
// If no seeds were specified, add the preferred seeds.
|
|
if self.seeds.is_empty() {
|
|
self.seeds = profile
|
|
.config
|
|
.preferred_seeds
|
|
.iter()
|
|
.map(|p| p.id)
|
|
.collect();
|
|
}
|
|
// Remove our local node from the seed set just in case it was added by mistake.
|
|
self.seeds.remove(profile.id());
|
|
self
|
|
}
|
|
}
|
|
|
|
impl Default for SyncSettings {
|
|
fn default() -> Self {
|
|
Self {
|
|
replicas: sync::ReplicationFactor::default(),
|
|
seeds: BTreeSet::new(),
|
|
timeout: DEFAULT_SYNC_TIMEOUT,
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Error while syncing.
|
|
#[derive(thiserror::Error, Debug)]
|
|
pub enum SyncError {
|
|
#[error(transparent)]
|
|
Repository(#[from] RepositoryError),
|
|
#[error(transparent)]
|
|
Node(#[from] radicle::node::Error),
|
|
#[error("all seeds timed out")]
|
|
AllSeedsTimedOut,
|
|
#[error(transparent)]
|
|
Target(#[from] sync::announce::TargetError),
|
|
}
|
|
|
|
impl SyncError {
|
|
fn is_connection_err(&self) -> bool {
|
|
match self {
|
|
Self::Node(e) => e.is_connection_err(),
|
|
_ => false,
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Writes sync output.
|
|
#[derive(Debug)]
|
|
pub enum SyncWriter {
|
|
/// Write to standard out.
|
|
Stdout(io::Stdout),
|
|
/// Write to standard error.
|
|
Stderr(io::Stderr),
|
|
/// Discard output, like [`std::io::sink`].
|
|
Sink,
|
|
}
|
|
|
|
impl Clone for SyncWriter {
|
|
fn clone(&self) -> Self {
|
|
match self {
|
|
Self::Stdout(_) => Self::Stdout(io::stdout()),
|
|
Self::Stderr(_) => Self::Stderr(io::stderr()),
|
|
Self::Sink => Self::Sink,
|
|
}
|
|
}
|
|
}
|
|
|
|
impl io::Write for SyncWriter {
|
|
fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
|
|
match self {
|
|
Self::Stdout(stdout) => stdout.write(buf),
|
|
Self::Stderr(stderr) => stderr.write(buf),
|
|
Self::Sink => Ok(buf.len()),
|
|
}
|
|
}
|
|
|
|
fn flush(&mut self) -> io::Result<()> {
|
|
match self {
|
|
Self::Stdout(stdout) => stdout.flush(),
|
|
Self::Stderr(stderr) => stderr.flush(),
|
|
Self::Sink => Ok(()),
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Configures how sync progress is reported.
|
|
pub struct SyncReporting {
|
|
/// Progress messages or animations.
|
|
pub progress: SyncWriter,
|
|
/// Completion messages.
|
|
pub completion: SyncWriter,
|
|
/// Debug output.
|
|
pub debug: bool,
|
|
}
|
|
|
|
impl Default for SyncReporting {
|
|
fn default() -> Self {
|
|
Self {
|
|
progress: SyncWriter::Stderr(io::stderr()),
|
|
completion: SyncWriter::Stdout(io::stdout()),
|
|
debug: false,
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Announce changes to the network.
|
|
pub fn announce<R: ReadRepository>(
|
|
repo: &R,
|
|
settings: SyncSettings,
|
|
reporting: SyncReporting,
|
|
node: &mut Node,
|
|
profile: &Profile,
|
|
) -> Result<Option<sync::AnnouncerResult>, SyncError> {
|
|
match announce_(repo, settings, reporting, node, profile) {
|
|
Ok(result) => Ok(result),
|
|
Err(e) if e.is_connection_err() => {
|
|
term::hint("Node is stopped. To announce changes to the network, start it with `rad node start`.");
|
|
Ok(None)
|
|
}
|
|
Err(e) => Err(e),
|
|
}
|
|
}
|
|
|
|
fn announce_<R>(
|
|
repo: &R,
|
|
settings: SyncSettings,
|
|
mut reporting: SyncReporting,
|
|
node: &mut Node,
|
|
profile: &Profile,
|
|
) -> Result<Option<sync::AnnouncerResult>, SyncError>
|
|
where
|
|
R: ReadRepository,
|
|
{
|
|
let me = profile.id();
|
|
let rid = repo.id();
|
|
let doc = repo.identity_doc()?;
|
|
|
|
let settings = settings.with_profile(profile);
|
|
let n_preferred_seeds = settings.seeds.len();
|
|
|
|
let config = match sync::PrivateNetwork::private_repo(&doc) {
|
|
None => {
|
|
let (synced, unsynced) = node.seeds(rid)?.iter().fold(
|
|
(BTreeSet::new(), BTreeSet::new()),
|
|
|(mut synced, mut unsynced), seed| {
|
|
if seed.is_synced() {
|
|
synced.insert(seed.nid);
|
|
} else {
|
|
unsynced.insert(seed.nid);
|
|
}
|
|
(synced, unsynced)
|
|
},
|
|
);
|
|
sync::AnnouncerConfig::public(*me, settings.replicas, settings.seeds, synced, unsynced)
|
|
}
|
|
Some(network) => {
|
|
let sessions = node.sessions()?;
|
|
let network =
|
|
network.restrict(|nid| sessions.iter().any(|s| s.nid == *nid && s.is_connected()));
|
|
sync::AnnouncerConfig::private(*me, settings.replicas, network)
|
|
}
|
|
};
|
|
let announcer = match sync::Announcer::new(config) {
|
|
Ok(announcer) => announcer,
|
|
Err(err) => match err {
|
|
sync::AnnouncerError::AlreadySynced(result) => {
|
|
term::success!(
|
|
&mut reporting.completion;
|
|
"Nothing to announce, already in sync with {} seed(s) (see `rad sync status`)",
|
|
term::format::positive(result.synced()),
|
|
);
|
|
return Ok(None);
|
|
}
|
|
sync::AnnouncerError::NoSeeds => {
|
|
term::info!(
|
|
&mut reporting.completion;
|
|
"{}",
|
|
term::format::yellow(format!("No seeds found for {rid}."))
|
|
);
|
|
return Ok(None);
|
|
}
|
|
sync::AnnouncerError::Target(err) => return Err(err.into()),
|
|
},
|
|
};
|
|
let target = announcer.target();
|
|
let min_replicas = target.replicas().lower_bound();
|
|
let mut spinner = term::spinner_to(
|
|
format!("Found {} seed(s)..", announcer.progress().unsynced()),
|
|
reporting.completion.clone(),
|
|
reporting.progress.clone(),
|
|
);
|
|
|
|
match node.announce(rid, settings.timeout, announcer, |node, progress| {
|
|
spinner.message(format!(
|
|
"Synced with {}, {} of {} preferred seeds, and {} of at least {} replica(s).",
|
|
term::format::node_id_human_compact(node),
|
|
term::format::secondary(progress.preferred()),
|
|
term::format::secondary(n_preferred_seeds),
|
|
term::format::secondary(progress.synced()),
|
|
// N.b. the number of replicas could exceed the target if we're
|
|
// waiting for preferred seeds
|
|
term::format::secondary(min_replicas.max(progress.synced())),
|
|
));
|
|
}) {
|
|
Ok(result) => {
|
|
spinner.message(format!(
|
|
"Synced with {} seed(s)",
|
|
term::format::positive(result.synced().len())
|
|
));
|
|
spinner.finish();
|
|
Ok(Some(result))
|
|
}
|
|
Err(err) => {
|
|
spinner.error(format!("Sync failed: {err}"));
|
|
Err(err.into())
|
|
}
|
|
}
|
|
}
|