diff --git a/radicle-cli/src/commands/clone.rs b/radicle-cli/src/commands/clone.rs index 6198fee9..f5d6eb05 100644 --- a/radicle-cli/src/commands/clone.rs +++ b/radicle-cli/src/commands/clone.rs @@ -10,7 +10,7 @@ use radicle::git::raw; use radicle::identity::doc; use radicle::identity::doc::{DocError, Id}; use radicle::node; -use radicle::node::{FetchLookup, Handle}; +use radicle::node::{FetchResult, Handle as _, Node}; use radicle::prelude::*; use radicle::rad; use radicle::storage; @@ -116,11 +116,9 @@ pub fn run(options: Options, ctx: impl term::Context) -> anyhow::Result<()> { } #[derive(Error, Debug)] -pub enum CloneError { +pub enum CloneError { #[error("node: {0}")] Node(#[from] node::Error), - #[error("fetch: {0}")] - Fetch(#[from] node::FetchError), #[error("fork: {0}")] Fork(#[from] rad::ForkError), #[error("storage: {0}")] @@ -133,50 +131,45 @@ pub enum CloneError { Payload(#[from] doc::PayloadError), #[error("project error: {0}")] Project(#[from] ProjectError), - #[error("handle error: {0}")] - Handle(H::Error), } -pub fn clone( +pub fn clone( id: Id, signer: &G, storage: &Storage, - node: &mut H, -) -> Result<(raw::Repository, Doc, Project), CloneError> { + node: &mut Node, +) -> Result<(raw::Repository, Doc, Project), CloneError> { let me = *signer.public_key(); - // Track & fetch project. - if node.track_repo(id).map_err(CloneError::Handle)? { + // Track. + if node.track_repo(id)? { term::success!( "Tracking relationship restablished for {}", term::format::tertiary(id) ); } - let spinner = term::spinner(format!("Fetching {}..", term::format::tertiary(id))); - match node.fetch(id).map_err(CloneError::Handle)? { - FetchLookup::Found { seeds, results } => { - // TODO: If none of them succeeds, output an error. Otherwise tell the caller - // how many succeeded. - for result in results.iter().take(seeds.len()) { - match &*result { - Ok(_updates) => {} - Err(_err) => {} - } + // Get seeds. + let seeds = node.seeds(id)?; + // Fetch from all seeds. + for seed in seeds { + let spinner = term::spinner(format!( + "Fetching {} from {}..", + term::format::tertiary(id), + term::format::tertiary(term::format::node(&seed)) + )); + + // TODO: If none of them succeeds, output an error. Otherwise tell the caller + // how many succeeded. + match node.fetch(id, seed)? { + FetchResult::Success { .. } => { + spinner.finish(); + } + FetchResult::Failed { reason } => { + spinner.error(reason); } } - FetchLookup::NotFound => { - // TODO: Return error. - } - FetchLookup::NotTracking => { - // SAFETY: Since we track it above, this shouldn't trigger unless there's a bug. - panic!("clone: Repository is not tracked"); - } - FetchLookup::Error(err) => { - return Err(err.into()); - } } - spinner.finish(); // Create a local fork of the project, under our own id. { diff --git a/radicle-cli/src/commands/track.rs b/radicle-cli/src/commands/track.rs index 00e5a97d..613c46f8 100644 --- a/radicle-cli/src/commands/track.rs +++ b/radicle-cli/src/commands/track.rs @@ -117,7 +117,8 @@ pub fn run(options: Options, ctx: impl term::Context) -> anyhow::Result<()> { } if options.fetch { - node.fetch(rid)?; + // TODO: Run a proper fetch here. + term::warning("fetch after track is not yet supported"); } Ok(()) diff --git a/radicle-cli/src/terminal/spinner.rs b/radicle-cli/src/terminal/spinner.rs index 348f6fd9..d2ccd7e9 100644 --- a/radicle-cli/src/terminal/spinner.rs +++ b/radicle-cli/src/terminal/spinner.rs @@ -32,19 +32,19 @@ impl Spinner { self.set_failed(); } - pub fn error(self, err: anyhow::Error) -> anyhow::Error { - self.progress.finish_and_clear(); - term::eprintln(style("!!").red().reverse(), style(&err).red()); + pub fn error(mut self, msg: impl ToString) { + let msg = msg.to_string(); - err + self.message = format!("{} ({})", self.message, msg); + self.set_failed(); } pub fn clear(self) { self.progress.finish_and_clear(); } - pub fn message(&mut self, msg: impl Into) { - let msg = msg.into(); + pub fn message(&mut self, msg: impl ToString) { + let msg = msg.to_string(); self.progress.set_message(msg.clone()); self.message = msg; diff --git a/radicle-node/src/control.rs b/radicle-node/src/control.rs index 93e23ac5..f3e56a93 100644 --- a/radicle-node/src/control.rs +++ b/radicle-node/src/control.rs @@ -8,10 +8,12 @@ use std::path::PathBuf; use std::{io, net}; use radicle::node::Handle; +use serde_json as json; use crate::identity::Id; use crate::node; -use crate::node::FetchLookup; +use crate::node::FetchResult; +use crate::node::NodeId; use crate::runtime; #[derive(thiserror::Error, Debug)] @@ -23,7 +25,7 @@ pub enum Error { } /// Listen for commands on the control socket, and process them. -pub fn listen>( +pub fn listen>( listener: UnixListener, mut handle: H, ) -> Result<(), Error> { @@ -34,8 +36,8 @@ pub fn listen>( Ok(mut stream) => { log::debug!(target: "control", "Accepted new client on control socket.."); - if let Err(e) = drain(&stream, &mut handle) { - if let DrainError::Shutdown = e { + if let Err(e) = command(&stream, &mut handle) { + if let CommandError::Shutdown = e { log::debug!(target: "control", "Shutdown requested.."); // Channel might already be disconnected if shutdown // came from somewhere else. Ignore errors. @@ -57,11 +59,13 @@ pub fn listen>( } #[derive(thiserror::Error, Debug)] -enum DrainError { +enum CommandError { #[error("invalid command argument `{0}`, {1}")] InvalidCommandArg(String, Box), #[error("unknown command `{0}`")] UnknownCommand(String), + #[error("serialization failed: {0}")] + Serialization(#[from] json::Error), #[error("runtime error: {0}")] Runtime(#[from] runtime::HandleError), #[error("i/o error: {0}")] @@ -70,10 +74,10 @@ enum DrainError { Shutdown, } -fn drain>( +fn command>( stream: &UnixStream, handle: &mut H, -) -> Result<(), DrainError> { +) -> Result<(), CommandError> { let mut reader = BufReader::new(stream); let mut writer = LineWriter::new(stream); let mut line = String::new(); @@ -86,14 +90,18 @@ fn drain>( // TODO: refactor to include helper match cmd.split_once(' ') { - Some(("fetch", arg)) => match arg.parse() { - Ok(id) => { - fetch(id, LineWriter::new(stream), handle)?; + Some(("fetch", args)) => { + if let Some((rid, node)) = args.split_once(' ') { + let rid: Id = rid + .parse() + .map_err(|e| CommandError::InvalidCommandArg(rid.to_owned(), Box::new(e)))?; + let node: NodeId = node + .parse() + .map_err(|e| CommandError::InvalidCommandArg(node.to_owned(), Box::new(e)))?; + + fetch(rid, node, LineWriter::new(stream), handle)?; } - Err(err) => { - return Err(DrainError::InvalidCommandArg(arg.to_owned(), Box::new(err))); - } - }, + } Some(("track-repo", arg)) => match arg.parse() { Ok(id) => match handle.track_repo(id) { Ok(updated) => { @@ -104,11 +112,14 @@ fn drain>( } } Err(e) => { - return Err(DrainError::Runtime(e)); + return Err(CommandError::Runtime(e)); } }, Err(err) => { - return Err(DrainError::InvalidCommandArg(arg.to_owned(), Box::new(err))); + return Err(CommandError::InvalidCommandArg( + arg.to_owned(), + Box::new(err), + )); } }, Some(("untrack-repo", arg)) => match arg.parse() { @@ -121,11 +132,14 @@ fn drain>( } } Err(e) => { - return Err(DrainError::Runtime(e)); + return Err(CommandError::Runtime(e)); } }, Err(err) => { - return Err(DrainError::InvalidCommandArg(arg.to_owned(), Box::new(err))); + return Err(CommandError::InvalidCommandArg( + arg.to_owned(), + Box::new(err), + )); } }, Some(("track-node", args)) => { @@ -144,11 +158,11 @@ fn drain>( } } Err(e) => { - return Err(DrainError::Runtime(e)); + return Err(CommandError::Runtime(e)); } }, Err(err) => { - return Err(DrainError::InvalidCommandArg( + return Err(CommandError::InvalidCommandArg( args.to_owned(), Box::new(err), )); @@ -165,30 +179,35 @@ fn drain>( } } Err(e) => { - return Err(DrainError::Runtime(e)); + return Err(CommandError::Runtime(e)); } }, Err(err) => { - return Err(DrainError::InvalidCommandArg(arg.to_owned(), Box::new(err))); + return Err(CommandError::InvalidCommandArg( + arg.to_owned(), + Box::new(err), + )); } }, Some(("announce-refs", arg)) => match arg.parse() { Ok(id) => { if let Err(e) = handle.announce_refs(id) { - return Err(DrainError::Runtime(e)); + return Err(CommandError::Runtime(e)); } writeln!(writer, "{}", node::RESPONSE_OK)?; } Err(err) => { - return Err(DrainError::InvalidCommandArg(arg.to_owned(), Box::new(err))); + return Err(CommandError::InvalidCommandArg( + arg.to_owned(), + Box::new(err), + )); } }, - Some((cmd, _)) => return Err(DrainError::UnknownCommand(cmd.to_owned())), + Some((cmd, _)) => return Err(CommandError::UnknownCommand(cmd.to_owned())), // Commands with no arguments. None => match cmd { "status" => { - println!("RECEIVED 'status'"); writeln!(writer, "{}", node::RESPONSE_OK).ok(); } "routing" => match handle.routing() { @@ -197,7 +216,7 @@ fn drain>( writeln!(writer, "{id} {seed}",)?; } } - Err(e) => return Err(DrainError::Runtime(e)), + Err(e) => return Err(CommandError::Runtime(e)), }, "inventory" => match handle.inventory() { Ok(c) => { @@ -205,68 +224,31 @@ fn drain>( writeln!(writer, "{id}")?; } } - Err(e) => return Err(DrainError::Runtime(e)), + Err(e) => return Err(CommandError::Runtime(e)), }, "shutdown" => { - return Err(DrainError::Shutdown); + return Err(CommandError::Shutdown); } _ => { - return Err(DrainError::UnknownCommand(line)); + return Err(CommandError::UnknownCommand(line)); } }, } Ok(()) } -fn fetch>( +fn fetch>( id: Id, + node: NodeId, mut writer: W, handle: &mut H, -) -> Result<(), DrainError> { - match handle.fetch(id) { +) -> Result<(), CommandError> { + match handle.fetch(id, node) { + Ok(result) => { + json::to_writer(&mut writer, &result)?; + } Err(e) => { - return Err(DrainError::Runtime(e)); - } - Ok(FetchLookup::Found { seeds, results }) => { - let seeds = Vec::from(seeds); - - writeln!( - writer, - "ok: found {} seeds for {} ({:?})", // TODO: Better output - seeds.len(), - &id, - &seeds, - )?; - - for result in results - .iter() - .take(results.capacity().unwrap_or(seeds.len())) - { - match result.result { - Ok(updated) => { - writeln!(writer, "ok: {id} fetched from {}", result.remote)?; - for update in updated { - writeln!(writer, "{update}")?; - } - } - Err(err) => { - writeln!( - writer, - "error: {id} failed to fetch from {}: {err}", - result.remote - )?; - } - } - } - } - Ok(FetchLookup::NotFound) => { - writeln!(writer, "error: {id} was not found")?; - } - Ok(FetchLookup::NotTracking) => { - writeln!(writer, "error: {id} is not tracked")?; - } - Ok(FetchLookup::Error(err)) => { - writeln!(writer, "error: {err}")?; + return Err(CommandError::Runtime(e)); } } Ok(()) diff --git a/radicle-node/src/runtime/handle.rs b/radicle-node/src/runtime/handle.rs index 83483a6b..1d0a79b9 100644 --- a/radicle-node/src/runtime/handle.rs +++ b/radicle-node/src/runtime/handle.rs @@ -10,7 +10,7 @@ use thiserror::Error; use crate::crypto::Signer; use crate::identity::Id; -use crate::node::FetchLookup; +use crate::node::FetchResult; use crate::profile::Home; use crate::service; use crate::service::{CommandError, QueryState}; @@ -107,6 +107,7 @@ impl Handle { impl radicle::node::Handle for Handle { type Sessions = Sessions; type Error = Error; + type FetchResult = FetchResult; fn is_running(&self) -> bool { true @@ -118,9 +119,15 @@ impl radicle::node::Handle for Handle { Ok(()) } - fn fetch(&mut self, id: Id) -> Result { + fn seeds(&mut self, id: Id) -> Result, Self::Error> { let (sender, receiver) = chan::bounded(1); - self.command(service::Command::Fetch(id, sender))?; + self.command(service::Command::Seeds(id, sender))?; + receiver.recv().map_err(Error::from) + } + + fn fetch(&mut self, id: Id, from: NodeId) -> Result { + let (sender, receiver) = chan::bounded(1); + self.command(service::Command::Fetch(id, from, sender))?; receiver.recv().map_err(Error::from) } diff --git a/radicle-node/src/service.rs b/radicle-node/src/service.rs index cd9a1154..b6005872 100644 --- a/radicle-node/src/service.rs +++ b/radicle-node/src/service.rs @@ -18,7 +18,6 @@ use crossbeam_channel as chan; use fastrand::Rng; use localtime::{LocalDuration, LocalTime}; use log::*; -use nonempty::NonEmpty; use crate::address; use crate::address::AddressBook; @@ -27,7 +26,7 @@ use crate::crypto; use crate::crypto::{Signer, Verified}; use crate::identity::{Doc, Id}; use crate::node; -use crate::node::{Address, Features, FetchError, FetchLookup, FetchResult}; +use crate::node::{Address, Features}; use crate::prelude::*; use crate::service::message::{Announcement, AnnouncementMessage, Ping}; use crate::service::message::{NodeAnnouncement, RefsAnnouncement}; @@ -35,6 +34,8 @@ use crate::service::session::Protocol; use crate::storage; use crate::storage::{Inventory, ReadRepository, RefUpdate, WriteStorage}; use crate::storage::{Namespaces, ReadStorage}; +use crate::worker; +use crate::worker::FetchError; use crate::Link; pub use crate::node::NodeId; @@ -105,8 +106,10 @@ pub enum Command { AnnounceRefs(Id), /// Connect to node with the given address. Connect(NodeId, Address), + /// Lookup seeds for the given repository in the routing table. + Seeds(Id, chan::Sender>), /// Fetch the given repository from the network. - Fetch(Id, chan::Sender), + Fetch(Id, NodeId, chan::Sender), /// Track the given repository. TrackRepo(Id, chan::Sender), /// Untrack the given repository. @@ -124,7 +127,8 @@ impl fmt::Debug for Command { match self { Self::AnnounceRefs(id) => write!(f, "AnnounceRefs({id})"), Self::Connect(id, addr) => write!(f, "Connect({id}, {addr})"), - Self::Fetch(id, _) => write!(f, "Fetch({id})"), + Self::Seeds(id, _) => write!(f, "Seeds({id})"), + Self::Fetch(id, node, _) => write!(f, "Fetch({id}, {node})"), Self::TrackRepo(id, _) => write!(f, "TrackRepo({id})"), Self::UntrackRepo(id, _) => write!(f, "UntrackRepo({id})"), Self::TrackNode(id, _, _) => write!(f, "TrackNode({id})"), @@ -172,7 +176,7 @@ pub struct Service { /// Whether our local inventory no long represents what we have announced to the network. out_of_sync: bool, /// Fetch requests initiated by user, which are waiting for results. - fetch_reqs: HashMap>, + fetch_reqs: HashMap>, /// Current tracked repository bloom filter. filter: Filter, /// Last time the service was idle. @@ -394,16 +398,7 @@ where match cmd { Command::Connect(id, addr) => self.reactor.connect(id, addr), - Command::Fetch(rid, resp) => { - if !self - .tracking - .is_repo_tracked(&rid) - .expect("Service::command: error accessing tracking configuration") - { - resp.send(FetchLookup::NotTracking).ok(); - return; - } - + Command::Seeds(rid, resp) => { let (connected, unconnected) = match self.routing.get(&rid) { Ok(seeds) => seeds .into_iter() @@ -411,47 +406,30 @@ where .partition::, _>(|node| self.sessions.is_negotiated(node)), Err(err) => { error!(target: "service", "Error reading routing table for {rid}: {err}"); - resp.send(FetchLookup::NotFound).ok(); + drop(resp); return; } }; - debug!( target: "service", "Found {} connected seed(s) and {} unconnected seed(s) for {}", connected.len(), unconnected.len(), rid ); - - let Some(seeds) = NonEmpty::from_vec(connected) else { - warn!(target: "service", "No connected seeds found for {}", rid); - resp.send(FetchLookup::NotFound).ok(); - - // TODO: Establish connections to unconnected seeds, and retry. - // TODO: Fetch requests should be queued and re-checked to see if they can - // be fulfilled everytime a new node connects. - return; - }; - - let (results_send, results) = chan::bounded(seeds.len()); - resp.send(FetchLookup::Found { - seeds: seeds.clone(), - results, - }) - .ok(); - - self.fetch_reqs.insert(rid, results_send); - - // TODO: Limit the number of seeds we fetch from? Randomize? - for seed in seeds { - self.fetch(rid, &seed); - } + resp.send(connected).ok(); + } + Command::Fetch(rid, seed, resp) => { + // TODO: Establish connections to unconnected seeds, and retry. + // TODO: Fetch requests should be queued and re-checked to see if they can + // be fulfilled everytime a new node connects. + self.fetch_reqs.insert(rid, resp); + self.fetch(rid, &seed); } Command::TrackRepo(id, resp) => { let tracked = self .track_repo(&id, tracking::Scope::All) .expect("Service::command: error tracking repository"); - // TODO: Try to fetch project if we weren't tracking it. + // TODO: Try to fetch project if we weren't tracking it before. resp.send(tracked).ok(); } Command::UntrackRepo(id, resp) => { @@ -508,11 +486,10 @@ where } } - pub fn fetched(&mut self, result: FetchResult) { - let remote = result.remote; - let rid = result.rid; - let namespaces = result.namespaces; - let initiated = result.initiated; + pub fn fetched(&mut self, result: worker::FetchResult) { + let remote = result.fetch.remote; + let rid = result.fetch.rid; + let initiated = result.fetch.initiated; if initiated { log::debug!( @@ -533,7 +510,7 @@ where if let FetchError::Io(_) = err { self.reactor - .disconnect(result.remote, DisconnectReason::Fetch(err)); + .disconnect(remote, DisconnectReason::Fetch(err)); return; } else { Err(err) @@ -544,16 +521,7 @@ where if let Some(results) = self.fetch_reqs.get(&rid) { log::debug!(target: "service", "Found existing fetch request, sending result.."); - if results - .send(FetchResult { - rid, - initiated, - remote, - namespaces, - result, - }) - .is_err() - { + if results.send(node::FetchResult::from(result)).is_err() { log::error!(target: "service", "Error sending fetch result for {rid}.."); self.fetch_reqs.remove(&rid); } else { diff --git a/radicle-node/src/service/reactor.rs b/radicle-node/src/service/reactor.rs index ebda4541..053913a4 100644 --- a/radicle-node/src/service/reactor.rs +++ b/radicle-node/src/service/reactor.rs @@ -29,7 +29,7 @@ pub enum Io { #[derive(Debug, Clone)] pub struct Fetch { /// Repo to fetch. - pub repo: Id, + pub rid: Id, /// Namespaces to fetch. pub namespaces: Namespaces, /// Remote peer we are interacting with. @@ -87,9 +87,9 @@ impl Reactor { self.io.push_back(Io::Wakeup(after)); } - pub fn fetch(&mut self, remote: NodeId, repo: Id, namespaces: Namespaces, initiated: bool) { + pub fn fetch(&mut self, remote: NodeId, rid: Id, namespaces: Namespaces, initiated: bool) { self.io.push_back(Io::Fetch(Fetch { - repo, + rid, namespaces, remote, initiated, diff --git a/radicle-node/src/test/handle.rs b/radicle-node/src/test/handle.rs index 56e48ee8..2581bdde 100644 --- a/radicle-node/src/test/handle.rs +++ b/radicle-node/src/test/handle.rs @@ -4,10 +4,11 @@ use std::sync::{Arc, Mutex}; use crossbeam_channel as chan; use crate::identity::Id; -use crate::node::FetchLookup; +use crate::node::FetchResult; use crate::runtime::HandleError; use crate::service; use crate::service::NodeId; +use crate::storage::RefUpdate; #[derive(Default, Clone)] pub struct Handle { @@ -19,6 +20,7 @@ pub struct Handle { impl radicle::node::Handle for Handle { type Error = HandleError; type Sessions = service::Sessions; + type FetchResult = FetchResult; fn is_running(&self) -> bool { true @@ -28,8 +30,12 @@ impl radicle::node::Handle for Handle { unimplemented!(); } - fn fetch(&mut self, _id: Id) -> Result { - Ok(FetchLookup::NotFound) + fn seeds(&mut self, _id: Id) -> Result, Self::Error> { + unimplemented!(); + } + + fn fetch(&mut self, _id: Id, _from: NodeId) -> Result { + Ok(FetchResult::from(Ok::, Self::Error>(vec![]))) } fn track_repo(&mut self, id: Id) -> Result { diff --git a/radicle-node/src/test/simulator.rs b/radicle-node/src/test/simulator.rs index eb4dbb3b..ea2e244b 100644 --- a/radicle-node/src/test/simulator.rs +++ b/radicle-node/src/test/simulator.rs @@ -14,13 +14,13 @@ use log::*; use crate::crypto::Signer; use crate::git::raw as git; -use crate::node::{FetchError, FetchResult}; use crate::prelude::Address; use crate::service::reactor::Io; use crate::service::{DisconnectReason, Event, Message, NodeId}; use crate::storage::{Namespaces, RefUpdate}; use crate::storage::{WriteRepository, WriteStorage}; use crate::test::peer::Service; +use crate::worker::{FetchError, FetchResult}; use crate::Link; /// Minimum latency between peers. @@ -112,7 +112,7 @@ impl fmt::Display for Scheduled { write!( f, "{} <~ {} ({}): FetchCompleted", - self.node, result.remote, result.rid + self.node, result.fetch.remote, result.fetch.rid ) } } @@ -413,9 +413,14 @@ impl Simulation { } Input::Fetched(result) => { let result = Arc::try_unwrap(result).unwrap(); - let mut repo = p.storage().repository(result.rid).unwrap(); + let mut repo = p.storage().repository(result.fetch.rid).unwrap(); - fetch(&mut repo, &result.remote, result.namespaces.clone()).unwrap(); + fetch( + &mut repo, + &result.fetch.remote, + result.fetch.namespaces.clone(), + ) + .unwrap(); p.fetched(result); } } @@ -617,10 +622,7 @@ impl Simulation { node, remote: fetch.remote, input: Input::Fetched(Arc::new(FetchResult { - rid: fetch.repo, - initiated: fetch.initiated, - remote: fetch.remote, - namespaces: fetch.namespaces, + fetch, result: Err(FetchError::Io(io::ErrorKind::Other.into())), })), }, @@ -632,10 +634,7 @@ impl Simulation { node, remote: fetch.remote, input: Input::Fetched(Arc::new(FetchResult { - rid: fetch.repo, - initiated: fetch.initiated, - remote: fetch.remote, - namespaces: fetch.namespaces, + fetch, result: Ok(vec![]), })), }, diff --git a/radicle-node/src/tests/e2e.rs b/radicle-node/src/tests/e2e.rs index cf9b246e..3c11d322 100644 --- a/radicle-node/src/tests/e2e.rs +++ b/radicle-node/src/tests/e2e.rs @@ -10,8 +10,7 @@ use radicle::crypto::test::signer::MockSigner; use radicle::crypto::Signer; use radicle::git::refname; use radicle::identity::Id; -use radicle::node::FetchLookup; -use radicle::node::Handle as _; +use radicle::node::{FetchResult, Handle as _}; use radicle::profile::Home; use radicle::storage::{ReadRepository, ReadStorage, WriteStorage}; use radicle::test::fixtures; @@ -348,23 +347,21 @@ fn test_replication() { let tracked = alice.handle.track_repo(acme).unwrap(); assert!(tracked); - let (seeds, results) = match alice.handle.fetch(acme).unwrap() { - FetchLookup::Found { seeds, results } => (seeds, results), - other => panic!("Fetch lookup failed, got {other:?}"), - }; - assert_eq!(seeds, nonempty::NonEmpty::new(bob.id)); + let seeds = alice.handle.seeds(acme).unwrap(); + assert!(seeds.contains(&bob.id)); - let result = results.recv_timeout(Duration::from_secs(6)).unwrap(); - let updated = match result.result { - Ok(updated) => updated, - Err(err) => { - panic!("Fetch failed from {}: {err}", result.remote); + let result = alice.handle.fetch(acme, bob.id).unwrap(); + assert!(result.is_success()); + + let updated = match result { + FetchResult::Success { updated } => updated, + FetchResult::Failed { reason } => { + panic!("Fetch failed from {}: {reason}", bob.id); } }; - assert_eq!(result.remote, bob.id); - assert_eq!(updated, vec![]); + assert_eq!(*updated, vec![]); - log::debug!(target: "test", "Fetch complete with {}", result.remote); + log::debug!(target: "test", "Fetch complete with {}", bob.id); let inventory = alice.handle.inventory().unwrap(); let alice_repo = alice.storage.repository(acme).unwrap(); @@ -404,13 +401,12 @@ fn test_clone() { transport::local::register(alice.storage.clone()); let _ = alice.handle.track_repo(acme).unwrap(); - let lookup = alice.handle.fetch(acme).unwrap(); + let seeds = alice.handle.seeds(acme).unwrap(); + assert!(seeds.contains(&bob.id)); + + let result = alice.handle.fetch(acme, bob.id).unwrap(); + assert!(result.is_success()); - match lookup { - // Drain the channel. - FetchLookup::Found { seeds, results } => for _ in results.iter().take(seeds.len()) {}, - other => panic!("Unexpected fetch lookup: {other:?}"), - } rad::fork(acme, &alice.signer, &alice.storage).unwrap(); let working = rad::checkout( @@ -455,15 +451,10 @@ fn test_fetch_up_to_date() { transport::local::register(alice.storage.clone()); let _ = alice.handle.track_repo(acme).unwrap(); - - match alice.handle.fetch(acme).unwrap() { - FetchLookup::Found { seeds, results } => for _ in results.iter().take(seeds.len()) {}, - other => panic!("Unexpected fetch lookup: {other:?}"), - } + let result = alice.handle.fetch(acme, bob.id).unwrap(); + assert!(result.is_success()); // Fetch again! This time, everything's up to date. - match alice.handle.fetch(acme).unwrap() { - FetchLookup::Found { seeds, results } => for _ in results.iter().take(seeds.len()) {}, - other => panic!("Unexpected fetch lookup: {other:?}"), - } + let result = alice.handle.fetch(acme, bob.id).unwrap(); + assert_eq!(result.success(), Some(vec![])); } diff --git a/radicle-node/src/wire/protocol.rs b/radicle-node/src/wire/protocol.rs index ab894a0a..5d8a43cf 100644 --- a/radicle-node/src/wire/protocol.rs +++ b/radicle-node/src/wire/protocol.rs @@ -156,7 +156,7 @@ impl Peer { fn upgraded(&mut self) -> Fetch { if let Self::Upgrading { fetch, id, link } = self { let fetch = fetch.clone(); - log::debug!(target: "wire", "Peer {id} upgraded for fetch {}", fetch.repo); + log::debug!(target: "wire", "Peer {id} upgraded for fetch {}", fetch.rid); *self = Self::Upgraded { id: *id, diff --git a/radicle-node/src/worker.rs b/radicle-node/src/worker.rs index abf604dd..c7fba61e 100644 --- a/radicle-node/src/worker.rs +++ b/radicle-node/src/worker.rs @@ -1,4 +1,5 @@ use std::io::{prelude::*, BufReader}; +use std::ops::Deref; use std::thread::JoinHandle; use std::{env, io, net, process, thread, time}; @@ -13,9 +14,9 @@ use radicle::storage::{Namespaces, ReadRepository, RefUpdate, WriteRepository, W use radicle::{git, Storage}; use reactor::poller::popol; -use crate::node::{FetchError, FetchResult}; use crate::runtime::Handle; use crate::service::reactor::Fetch; +use crate::storage; use crate::wire::{WireReader, WireSession, WireWriter}; /// Worker pool configuration. @@ -34,6 +35,36 @@ pub struct Config { pub storage: Storage, } +/// Result of a fetch request from a specific seed. +#[derive(Debug)] +pub struct FetchResult { + pub fetch: Fetch, + pub result: Result, FetchError>, +} + +impl Deref for FetchResult { + type Target = Result, FetchError>; + + fn deref(&self) -> &Self::Target { + &self.result + } +} + +/// Error returned by fetch. +#[derive(thiserror::Error, Debug)] +pub enum FetchError { + #[error(transparent)] + Git(#[from] git::raw::Error), + #[error(transparent)] + Storage(#[from] storage::Error), + #[error(transparent)] + Fetch(#[from] storage::FetchError), + #[error(transparent)] + Io(#[from] io::Error), + #[error(transparent)] + Project(#[from] storage::ProjectError), +} + /// Task to be accomplished on a worker thread. /// This is either going to be an outgoing or incoming fetch. pub struct Task { @@ -77,13 +108,7 @@ impl Worker { } = task; let (session, result) = self._process(&fetch, drain, session); - let result = FetchResult { - rid: fetch.repo, - remote: fetch.remote, - namespaces: fetch.namespaces, - initiated: fetch.initiated, - result, - }; + let result = FetchResult { fetch, result }; log::debug!(target: "worker", "Sending response back to service.."); if self @@ -102,7 +127,7 @@ impl Worker { mut session: WireSession, ) -> (WireSession, Result, FetchError>) { if fetch.initiated { - log::debug!(target: "worker", "Worker processing outgoing fetch for {}", fetch.repo); + log::debug!(target: "worker", "Worker processing outgoing fetch for {}", fetch.rid); let mut tunnel = match Tunnel::with(session, net::SocketAddr::from(([0, 0, 0, 0], 0))) { Ok(tunnel) => tunnel, @@ -122,7 +147,7 @@ impl Worker { } (session, result) } else { - log::debug!(target: "worker", "Worker processing incoming fetch for {}", fetch.repo); + log::debug!(target: "worker", "Worker processing incoming fetch for {}", fetch.rid); if let Err(err) = session.as_connection_mut().set_nonblocking(false) { return (session, Err(err.into())); @@ -148,7 +173,7 @@ impl Worker { fetch: &Fetch, tunnel: &mut Tunnel>, ) -> Result, FetchError> { - let repo = self.storage.repository(fetch.repo)?; + let repo = self.storage.repository(fetch.rid)?; let tunnel_addr = tunnel.local_addr()?; let mut cmd = process::Command::new("git"); cmd.current_dir(repo.path()) @@ -197,13 +222,13 @@ impl Worker { // TODO: Parse fetch output to return updates. if child.wait()?.success() { - log::debug!(target: "worker", "Fetch for {} exited successfully", fetch.repo); + log::debug!(target: "worker", "Fetch for {} exited successfully", fetch.rid); } else { - log::error!(target: "worker", "Fetch for {} failed", fetch.repo); + log::error!(target: "worker", "Fetch for {} failed", fetch.rid); } let head = repo.set_head()?; - log::debug!(target: "worker", "Head for {} set to {head}", fetch.repo); + log::debug!(target: "worker", "Head for {} set to {head}", fetch.rid); Ok(vec![]) } @@ -228,9 +253,9 @@ impl Worker { Ok((req, pktline)) => { log::debug!( target: "worker", - "Parsed git command packet-line for {}: {:?}", fetch.repo, req + "Parsed git command packet-line for {}: {:?}", fetch.rid, req ); - if req.repo != fetch.repo { + if req.repo != fetch.rid { return Err(FetchError::Git(git::raw::Error::from_str( "git pkt-line command does not match fetch request", ))); @@ -254,17 +279,17 @@ impl Worker { if e.kind() == io::ErrorKind::UnexpectedEof { break; } - log::debug!(target: "worker", "Upload of {} to {} returned error: {e}", fetch.repo, fetch.remote); + log::debug!(target: "worker", "Upload of {} to {} returned error: {e}", fetch.rid, fetch.remote); return Err(e.into()); } if let Err(e) = stream_r.read_pktlines(&mut daemon_w, &mut buffer) { - log::error!(target: "worker", "Remote returned error for {}: {e}", fetch.repo); + log::error!(target: "worker", "Remote returned error for {}: {e}", fetch.rid); return Err(e.into()); } } - log::debug!(target: "worker", "Upload of {} to {} exited successfully", fetch.repo, fetch.remote); + log::debug!(target: "worker", "Upload of {} to {} exited successfully", fetch.rid, fetch.remote); // When we aren't the one fetching, no refs are updated. Ok(vec![]) diff --git a/radicle/src/node.rs b/radicle/src/node.rs index 2fe9c1b4..da69778e 100644 --- a/radicle/src/node.rs +++ b/radicle/src/node.rs @@ -1,21 +1,20 @@ mod features; -use amplify::WrapperMut; use std::io::{BufRead, BufReader, Write}; -use std::ops::Deref; use std::os::unix::net::UnixStream; use std::path::{Path, PathBuf}; +use std::str::FromStr; use std::{io, net}; +use amplify::WrapperMut; use crossbeam_channel as chan; use cyphernet::addr::{HostName, NetAddr}; -use nonempty::NonEmpty; +use serde::{Deserialize, Serialize}; +use serde_json as json; use crate::crypto::PublicKey; -use crate::git; use crate::identity::Id; -use crate::storage; -use crate::storage::{Namespaces, RefUpdate}; +use crate::storage::RefUpdate; pub use features::Features; @@ -55,55 +54,35 @@ impl From for Address { } } -/// Result of a fetch request from a specific seed. -#[derive(Debug)] -#[allow(clippy::large_enum_variant)] -pub struct FetchResult { - pub rid: Id, - pub remote: NodeId, - pub initiated: bool, - pub namespaces: Namespaces, - pub result: Result, FetchError>, +#[derive(Debug, Serialize, Deserialize)] +#[serde(tag = "status", rename_all = "kebab-case")] +pub enum FetchResult { + Success { updated: Vec }, + Failed { reason: String }, } -impl Deref for FetchResult { - type Target = Result, FetchError>; +impl FetchResult { + pub fn is_success(&self) -> bool { + matches!(self, FetchResult::Success { .. }) + } - fn deref(&self) -> &Self::Target { - &self.result + pub fn success(self) -> Option> { + match self { + Self::Success { updated } => Some(updated), + _ => None, + } } } -/// Error returned by fetch. -#[derive(thiserror::Error, Debug)] -pub enum FetchError { - #[error(transparent)] - Git(#[from] git::raw::Error), - #[error(transparent)] - Storage(#[from] storage::Error), - #[error(transparent)] - Fetch(#[from] storage::FetchError), - #[error(transparent)] - Io(#[from] io::Error), - #[error(transparent)] - Project(#[from] storage::ProjectError), -} - -/// Result of looking up seeds in our routing table. -/// This object is sent back to the caller who initiated the fetch. -#[derive(Debug)] -pub enum FetchLookup { - /// Found seeds for the given project. - Found { - seeds: NonEmpty, - results: chan::Receiver, - }, - /// Can't fetch because no seeds were found for this project. - NotFound, - /// Can't fetch because the project isn't tracked. - NotTracking, - /// Error trying to find seeds. - Error(FetchError), +impl From, S>> for FetchResult { + fn from(value: Result, S>) -> Self { + match value { + Ok(updated) => Self::Success { updated }, + Err(err) => Self::Failed { + reason: err.to_string(), + }, + } + } } #[derive(thiserror::Error, Debug)] @@ -112,6 +91,12 @@ pub enum Error { Connect(#[from] io::Error), #[error("received invalid response for `{cmd}` command: '{response}'")] InvalidResponse { cmd: &'static str, response: String }, + #[error("received invalid json in response for `{cmd}` command: '{response}': {error}")] + InvalidJson { + cmd: &'static str, + response: String, + error: json::Error, + }, #[error("received empty response for `{cmd}` command")] EmptyResponse { cmd: &'static str }, } @@ -122,13 +107,17 @@ pub trait Handle { type Sessions; /// The error returned by all methods. type Error: std::error::Error + Send + Sync + 'static; + /// Result of a fetch. + type FetchResult; /// Check if the node is running. to a peer. fn is_running(&self) -> bool; /// Connect to a peer. fn connect(&mut self, node: NodeId, addr: Address) -> Result<(), Self::Error>; - /// Retrieve or update the project from network. - fn fetch(&mut self, id: Id) -> Result; + /// Lookup the seeds of a given repository in the routing table. + fn seeds(&mut self, id: Id) -> Result, Self::Error>; + /// Fetch a repository from the network. + fn fetch(&mut self, id: Id, from: NodeId) -> Result; /// Start tracking the given project. Doesn't do anything if the project is already /// tracked. fn track_repo(&mut self, id: Id) -> Result; @@ -192,6 +181,7 @@ impl Node { impl Handle for Node { type Sessions = (); type Error = Error; + type FetchResult = FetchResult; fn is_running(&self) -> bool { let Ok(mut lines) = self.call::<&str>("status", &[]) else { @@ -207,13 +197,31 @@ impl Handle for Node { todo!() } - fn fetch(&mut self, id: Id) -> Result { - for line in self.call("fetch", &[id.urn()])? { - let line = line?; - log::debug!("node: {}", line); - } - // TODO: Return parsed lookup results. - Ok(FetchLookup::NotFound) + fn seeds(&mut self, id: Id) -> Result, Error> { + self.call("seeds", &[id.urn()])? + .map(|line| { + let line = line?; + let node = NodeId::from_str(&line).map_err(|_| Error::InvalidResponse { + cmd: "seeds", + response: line, + })?; + Ok(node) + }) + .collect() + } + + fn fetch(&mut self, id: Id, from: NodeId) -> Result { + let result = self + .call("fetch", &[id.urn(), from.to_human()])? + .next() + .ok_or(Error::EmptyResponse { cmd: "fetch" })??; + let lookup = json::from_str(&result).map_err(|e| Error::InvalidJson { + cmd: "fetch", + response: result, + error: e, + })?; + + Ok(lookup) } fn track_node(&mut self, id: NodeId, alias: Option) -> Result { diff --git a/radicle/src/storage.rs b/radicle/src/storage.rs index 65faf962..002b0f9e 100644 --- a/radicle/src/storage.rs +++ b/radicle/src/storage.rs @@ -6,7 +6,7 @@ use std::ops::Deref; use std::path::Path; use std::{fmt, io}; -use serde::Serialize; +use serde::{Deserialize, Serialize}; use thiserror::Error; use crypto::{PublicKey, Signer, Unverified, Verified}; @@ -92,7 +92,8 @@ pub enum FetchError { pub type RemoteId = PublicKey; /// An update to a reference. -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "kebab-case")] pub enum RefUpdate { Updated { name: RefString, old: Oid, new: Oid }, Created { name: RefString, oid: Oid },