mab122
/
radicle-heartwood-lfs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Update `clone` functionality for new replication

This commit is contained in:
Alexis Sellier 2023-01-22 22:48:36 +01:00
parent ef950d515a
commit fea14ff7b3
No known key found for this signature in database
13 changed files with 238 additions and 109 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
radicle-node/src/client/handle.rs
View File

@ -1,3 +1,4 @@
use std::fmt;
use std::io::{self, Write};
use std::os::unix::net::UnixStream;
use std::sync::atomic::{AtomicBool, Ordering};
@ -9,9 +10,10 @@ use thiserror::Error;
use crate::crypto::Signer;
use crate::identity::Id;
use crate::node::FetchLookup;
use crate::profile::Home;
use crate::service;
use crate::service::{CommandError, FetchLookup, QueryState};
use crate::service::{CommandError, QueryState};
use crate::service::{NodeId, Sessions};
use crate::wire;
use crate::worker::WorkerResp;
@ -62,6 +64,12 @@ pub struct Handle<G: Signer + EcSign> {
shutdown: Arc<AtomicBool>,
}
impl<G: Signer + EcSign> fmt::Debug for Handle<G> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("Handle").field("home", &self.home).finish()
}
}
impl<G: Signer + EcSign> Clone for Handle<G> {
fn clone(&self) -> Self {
Self {
@ -98,7 +106,6 @@ impl<G: Signer + EcSign + 'static> Handle<G> {
impl<G: Signer + EcSign + 'static> radicle::node::Handle for Handle<G> {
type Sessions = Sessions;
type FetchLookup = FetchLookup;
type Error = Error;
fn connect(&mut self, node: NodeId, addr: radicle::node::Address) -> Result<(), Error> {
@ -107,7 +114,7 @@ impl<G: Signer + EcSign + 'static> radicle::node::Handle for Handle<G> {
Ok(())
}
fn fetch(&mut self, id: Id) -> Result<Self::FetchLookup, Error> {
fn fetch(&mut self, id: Id) -> Result<FetchLookup, Error> {
let (sender, receiver) = chan::bounded(1);
self.command(service::Command::Fetch(id, sender))?;
receiver.recv().map_err(Error::from)

12
radicle-node/src/control.rs
View File

@ -12,7 +12,7 @@ use radicle::node::Handle;
use crate::client;
use crate::identity::Id;
use crate::node;
use crate::service::FetchLookup;
use crate::node::FetchLookup;
#[derive(thiserror::Error, Debug)]
pub enum Error {
@ -23,10 +23,7 @@ pub enum Error {
}
/// Listen for commands on the control socket, and process them.
pub fn listen<
P: AsRef<Path>,
H: Handle<Error = client::handle::Error, FetchLookup = FetchLookup>,
>(
pub fn listen<P: AsRef<Path>, H: Handle<Error = client::handle::Error>>(
path: P,
mut handle: H,
) -> Result<(), Error> {
@ -41,6 +38,7 @@ pub fn listen<
log::info!("Binding control socket {}..", path.as_ref().display());
// TODO: Move socket binding to main thread.
let listener = UnixListener::bind(path).map_err(Error::Bind)?;
for incoming in listener.incoming() {
match incoming {
@ -85,7 +83,7 @@ enum DrainError {
Shutdown,
}
fn drain<H: Handle<Error = client::handle::Error, FetchLookup = FetchLookup>>(
fn drain<H: Handle<Error = client::handle::Error>>(
stream: &UnixStream,
handle: &mut H,
) -> Result<(), DrainError> {
@ -220,7 +218,7 @@ fn drain<H: Handle<Error = client::handle::Error, FetchLookup = FetchLookup>>(
Ok(())
}
fn fetch<W: Write, H: Handle<Error = client::handle::Error, FetchLookup = FetchLookup>>(
fn fetch<W: Write, H: Handle<Error = client::handle::Error>>(
id: Id,
mut writer: W,
handle: &mut H,

49
radicle-node/src/service.rs
View File

@ -12,30 +12,29 @@ use std::collections::hash_map::Entry;
use std::collections::{BTreeMap, HashMap, HashSet};
use std::ops::{Deref, DerefMut};
use std::sync::Arc;
use std::{fmt, io, net, str};
use std::{fmt, net, str};
use crossbeam_channel as chan;
use fastrand::Rng;
use localtime::{LocalDuration, LocalTime};
use log::*;
use nonempty::NonEmpty;
use radicle::node::{Address, Features};
use radicle::storage::{Namespaces, ReadStorage};
use crate::address;
use crate::address::AddressBook;
use crate::clock::Timestamp;
use crate::crypto;
use crate::crypto::{Signer, Verified};
use crate::git;
use crate::identity::{Doc, Id};
use crate::node;
use crate::node::{Address, Features, FetchLookup, FetchResult};
use crate::prelude::*;
use crate::service::message::{Announcement, AnnouncementMessage, Ping};
use crate::service::message::{NodeAnnouncement, RefsAnnouncement};
use crate::service::session::Protocol;
use crate::storage;
use crate::storage::{Inventory, ReadRepository, RefUpdate, WriteStorage};
use crate::storage::{Namespaces, ReadStorage};
use crate::Link;
pub use crate::node::NodeId;
@ -97,48 +96,6 @@ pub enum Error {
Routing(#[from] routing::Error),
}
/// Error returned by [`Command::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<NodeId>,
results: chan::Receiver<FetchResult>,
},
/// 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),
}
/// 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 namespaces: Namespaces,
pub result: Result<Vec<RefUpdate>, FetchError>,
}
/// Function used to query internal service state.
pub type QueryState = dyn Fn(&dyn ServiceState) -> Result<(), CommandError> + Send + Sync;

3
radicle-node/src/test/handle.rs
View File

@ -5,8 +5,8 @@ use crossbeam_channel as chan;
use crate::client::handle::Error;
use crate::identity::Id;
use crate::node::FetchLookup;
use crate::service;
use crate::service::FetchLookup;
use crate::service::NodeId;
#[derive(Default, Clone)]
@ -19,7 +19,6 @@ pub struct Handle {
impl radicle::node::Handle for Handle {
type Error = Error;
type Sessions = service::Sessions;
type FetchLookup = FetchLookup;
fn connect(&mut self, _node: NodeId, _addr: radicle::node::Address) -> Result<(), Error> {
unimplemented!();

2
radicle-node/src/test/simulator.rs
View File

@ -13,10 +13,10 @@ use localtime::{LocalDuration, LocalTime};
use log::*;
use crate::crypto::Signer;
use crate::node::{FetchError, FetchResult};
use crate::prelude::Address;
use crate::service::reactor::Io;
use crate::service::{DisconnectReason, Event, Message, NodeId};
use crate::service::{FetchError, FetchResult};
use crate::storage::{WriteRepository, WriteStorage};
use crate::test::peer::Service;
use crate::Link;

75
radicle-node/src/tests/e2e.rs
View File

@ -10,15 +10,15 @@ 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::profile::Home;
use radicle::storage::{ReadStorage, WriteStorage};
use radicle::storage::{ReadRepository, ReadStorage, WriteStorage};
use radicle::test::fixtures;
use radicle::Storage;
use radicle::{assert_matches, rad};
use crate::node::NodeId;
use crate::service::FetchLookup;
use crate::storage::git::transport;
use crate::test::logger;
use crate::{client, client::handle::Handle, client::Runtime, service};
@ -34,6 +34,7 @@ struct Node {
struct NodeHandle {
id: NodeId,
storage: Storage,
signer: MockSigner,
addr: net::SocketAddr,
thread: ManuallyDrop<thread::JoinHandle<Result<(), client::Error>>>,
handle: ManuallyDrop<Handle<MockSigner>>,
@ -117,6 +118,7 @@ impl Node {
NodeHandle {
id,
storage: self.storage,
signer: self.signer,
addr,
handle,
thread,
@ -355,27 +357,64 @@ fn test_replication() {
log::debug!(target: "test", "Fetch complete with {}", result.remote);
// TODO: Have simpler way of listing all refs.
let inventory = alice.handle.inventory().unwrap();
let alice_refs = alice
.storage
.repository(acme)
let alice_repo = alice.storage.repository(acme).unwrap();
let bob_repo = bob.storage.repository(acme).unwrap();
let alice_refs = alice_repo
.references()
.unwrap()
.remotes()
.collect::<Result<Vec<_>, _>>()
.unwrap();
let bob_refs = bob_repo
.references()
.unwrap()
.map(|r| r.unwrap())
.collect::<Vec<_>>();
let bob_refs = bob
.storage
.repository(acme)
.unwrap()
.remotes()
.unwrap()
.map(|r| r.unwrap())
.collect::<Vec<_>>();
.collect::<Result<Vec<_>, _>>()
.unwrap();
assert_eq!(inventory.try_iter().next(), Some(acme));
assert_eq!(alice_refs, bob_refs);
assert_matches!(alice.storage.repository(acme).unwrap().verify(), Ok(()));
}
#[test]
fn test_clone() {
logger::init(log::Level::Debug);
let tmp = tempfile::tempdir().unwrap();
let alice = Node::new(tmp.path());
let mut bob = Node::new(tmp.path());
let acme = bob.project("acme");
let mut alice = alice.spawn(service::Config::default());
let bob = bob.spawn(service::Config::default());
alice.connect(&bob);
converge([&alice, &bob]);
transport::local::register(alice.storage.clone());
let repo = rad::clone(
acme,
tmp.path().join("clone"),
&alice.signer,
&alice.storage,
&mut (*alice.handle),
)
.unwrap();
// Makes test finish faster.
drop(alice);
let head = repo.head().unwrap();
let oid = head.target().unwrap();
let (_, canonical) = bob
.storage
.repository(acme)
.unwrap()
.canonical_head()
.unwrap();
assert_eq!(oid, *canonical);
}

14
radicle-node/src/worker.rs
View File

@ -14,8 +14,8 @@ use radicle::{git, Storage};
use reactor::poller::popol;
use crate::client::handle::Handle;
use crate::node::{FetchError, FetchResult};
use crate::service::reactor::Fetch;
use crate::service::{FetchError, FetchResult};
use crate::wire::{WireReader, WireSession, WireWriter};
/// Worker request.
@ -70,7 +70,7 @@ impl<G: Signer + EcSign + 'static> Worker<G> {
.worker_result(WorkerResp { result, session })
.is_err()
{
log::error!("Unable to report fetch result: worker channel disconnected");
log::error!(target: "worker", "Unable to report fetch result: worker channel disconnected");
}
}
@ -144,9 +144,9 @@ impl<G: Signer + EcSign + 'static> Worker<G> {
log::debug!(target: "worker", "Fetch for {} exited with status {:?}", fetch.repo, status.code());
if let Some(status) = status.code() {
log::debug!(target: "worker", "Upload pack for {} exited with status {:?}", fetch.repo, status);
log::debug!(target: "worker", "Fetch for {} exited with status {:?}", fetch.repo, status);
} else {
log::debug!(target: "worker", "Upload pack for {} exited with unknown status", fetch.repo);
log::debug!(target: "worker", "Fetch for {} exited with unknown status", fetch.repo);
}
if !status.success() {
@ -221,9 +221,9 @@ impl<G: Signer + EcSign + 'static> Worker<G> {
let status = child.wait()?;
if let Some(status) = status.code() {
log::debug!(target: "worker", "Upload pack for {} exited with status {:?}", fetch.repo, status);
log::debug!(target: "worker", "Upload-pack for {} exited with status {:?}", fetch.repo, status);
} else {
log::debug!(target: "worker", "Upload pack for {} exited with unknown status", fetch.repo);
log::debug!(target: "worker", "Upload-pack for {} exited with unknown status", fetch.repo);
}
if !status.success() {
@ -231,7 +231,7 @@ impl<G: Signer + EcSign + 'static> Worker<G> {
stderr.read_to_end(&mut err)?;
let err = String::from_utf8_lossy(&err);
log::debug!(target: "worker", "Upload pack for {}: stderr: {}", fetch.repo, err);
log::debug!(target: "worker", "Upload-pack for {}: stderr: {}", fetch.repo, err);
}
Ok(vec![])

6
radicle/src/git.rs
View File

@ -41,7 +41,7 @@ pub enum RefError {
#[error("expected ref to begin with 'refs/namespaces' but found '{0}'")]
MissingNamespace(format::RefString),
#[error("ref name contains invalid namespace identifier '{name}'")]
Id {
InvalidNamespace {
name: format::RefString,
#[source]
err: Box<dyn std::error::Error + Send + Sync + 'static>,
@ -224,7 +224,7 @@ where
.namespace()
.as_str()
.parse()
.map_err(|err| RefError::Id {
.map_err(|err| RefError::InvalidNamespace {
name: input.to_owned(),
err: Box::new(err),
})?;
@ -300,7 +300,7 @@ pub fn configure_remote<'r>(
url: &Url,
) -> Result<git2::Remote<'r>, git2::Error> {
let fetch = format!("+refs/heads/*:refs/remotes/{name}/*");
let remote = repo.remote_with_fetch(name, url.to_string().as_str(), &fetch)?;
let remote = repo.remote_with_fetch(name, dbg!(url.to_string().as_str()), &fetch)?;
Ok(remote)
}

66
radicle/src/node.rs
View File

@ -2,15 +2,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;
use std::{io, net};
use crossbeam_channel as chan;
use cyphernet::addr::{HostName, NetAddr};
use nonempty::NonEmpty;
use crate::crypto::PublicKey;
use crate::git;
use crate::identity::Id;
use crossbeam_channel as chan;
use crate::storage;
use crate::storage::{Namespaces, RefUpdate};
pub use features::Features;
@ -50,6 +55,56 @@ impl From<net::SocketAddr> 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 namespaces: Namespaces,
pub result: Result<Vec<RefUpdate>, FetchError>,
}
impl Deref for FetchResult {
type Target = Result<Vec<RefUpdate>, 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),
}
/// 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<NodeId>,
results: chan::Receiver<FetchResult>,
},
/// 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),
}
#[derive(thiserror::Error, Debug)]
pub enum Error {
#[error("failed to connect to node: {0}")]
@ -62,8 +117,6 @@ pub enum Error {
/// A handle to send commands to the node or request information.
pub trait Handle {
/// The result of a fetch request.
type FetchLookup;
/// The peer sessions type.
type Sessions;
/// The error returned by all methods.
@ -72,7 +125,7 @@ pub trait Handle {
/// 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<Self::FetchLookup, Self::Error>;
fn fetch(&mut self, id: Id) -> Result<FetchLookup, Self::Error>;
/// Start tracking the given project. Doesn't do anything if the project is already
/// tracked.
fn track_repo(&mut self, id: Id) -> Result<bool, Self::Error>;
@ -130,19 +183,18 @@ impl Node {
impl Handle for Node {
type Sessions = ();
type FetchLookup = ();
type Error = Error;
fn connect(&mut self, _node: NodeId, _addr: Address) -> Result<(), Error> {
todo!()
}
fn fetch(&mut self, id: Id) -> Result<(), Error> {
fn fetch(&mut self, id: Id) -> Result<FetchLookup, Error> {
for line in self.call("fetch", &[id.urn()])? {
let line = line?;
log::debug!("node: {}", line);
}
Ok(())
todo!()
}
fn track_node(&mut self, id: NodeId, alias: Option<String>) -> Result<bool, Error> {

46
radicle/src/rad.rs
View File

@ -11,8 +11,8 @@ use crate::git;
use crate::identity::doc;
use crate::identity::doc::{DocError, Id};
use crate::identity::project::Project;
use crate::node;
use crate::node::NodeId;
use crate::node::{self, FetchLookup};
use crate::storage::git::transport::{self, remote};
use crate::storage::git::{ProjectError, Repository, Storage};
use crate::storage::refs::SignedRefs;
@ -193,29 +193,55 @@ pub fn fork<G: Signer, S: storage::WriteStorage>(
}
#[derive(Error, Debug)]
pub enum CloneError {
pub enum CloneError<H: node::Handle> {
#[error("node: {0}")]
Node(#[from] node::Error),
#[error("fetch: {0}")]
Fetch(#[from] node::FetchError),
#[error("fork: {0}")]
Fork(#[from] ForkError),
#[error("checkout: {0}")]
Checkout(#[from] CheckoutError),
#[error("identity document error: {0}")]
Doc(#[from] DocError),
#[error("handle error: {0}")]
Handle(H::Error),
}
pub fn clone<P: AsRef<Path>, G: Signer, S: storage::WriteStorage, H: node::Handle>(
pub fn clone<P: AsRef<Path>, G: Signer, H: node::Handle>(
proj: Id,
path: P,
signer: &G,
storage: &S,
storage: &Storage,
handle: &mut H,
) -> Result<git2::Repository, CloneError>
where
CloneError: From<H::Error>,
{
let _ = handle.track_repo(proj)?;
let _ = handle.fetch(proj)?;
) -> Result<git2::Repository, CloneError<H>> {
let _ = handle.track_repo(proj).map_err(CloneError::Handle)?;
let lookup = handle.fetch(proj).map_err(CloneError::Handle)?;
match lookup {
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) => {}
}
}
}
FetchLookup::NotFound => {
// TODO: Return error instead.
panic!("clone: Repository not found in routing table");
}
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());
}
}
let _ = fork(proj, signer, storage)?;
let working = checkout(proj, signer.public_key(), path, storage)?;

2
radicle/src/storage.rs
View File

@ -303,7 +303,7 @@ pub trait ReadRepository {
remote: &RemoteId,
reference: &Qualified,
) -> Result<Oid, git_ext::Error>;
fn references(&self, remote: &RemoteId) -> Result<Refs, Error>;
fn references_of(&self, remote: &RemoteId) -> Result<Refs, Error>;
fn remote(&self, remote: &RemoteId) -> Result<Remote<Verified>, refs::Error>;
fn remotes(&self) -> Result<Remotes<Verified>, refs::Error>;
/// Return the project associated with this repository.

57
radicle/src/storage/git.rs
View File

@ -30,6 +30,36 @@ pub static NAMESPACES_GLOB: Lazy<refspec::PatternString> =
pub static SIGREFS_GLOB: Lazy<refspec::PatternString> =
Lazy::new(|| refspec::pattern!("refs/namespaces/*/rad/sigrefs"));
/// A parsed Git reference.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Ref {
pub oid: git::Oid,
pub name: RefString,
pub namespace: Option<RemoteId>,
}
impl<'a> TryFrom<git2::Reference<'a>> for Ref {
type Error = RefError;
fn try_from(r: git2::Reference) -> Result<Self, Self::Error> {
let name = r.name().ok_or(RefError::InvalidName)?;
let (namespace, name) = match git::parse_ref_namespaced::<RemoteId>(name) {
Ok((namespace, refname)) => (Some(namespace), refname.to_ref_string()),
Err(RefError::MissingNamespace(refname)) => (None, refname),
Err(err) => return Err(err),
};
let Some(oid) = r.target() else {
// Ignore symbolic refs, eg. `HEAD`.
return Err(RefError::Symbolic(name));
};
Ok(Self {
namespace,
name,
oid: oid.into(),
})
}
}
// TODO: Is this is the wrong place for this type?
#[derive(Error, Debug)]
pub enum ProjectError {
@ -261,6 +291,27 @@ impl Repository {
Ok(())
}
/// Iterate over all references.
pub fn references(
&self,
) -> Result<impl Iterator<Item = Result<Ref, refs::Error>> + '_, git2::Error> {
let refs = self
.backend
.references()?
.map(|reference| {
let r = reference?;
match Ref::try_from(r) {
Err(RefError::Symbolic(_)) => Ok(None),
Err(err) => Err(err.into()),
Ok(r) => Ok(Some(r)),
}
})
.filter_map(Result::transpose);
Ok(refs)
}
pub fn identity(&self, remote: &RemoteId) -> Result<Identity<Oid>, IdentityError> {
Identity::load(remote, self)
}
@ -448,7 +499,7 @@ impl ReadRepository for Repository {
Ok(Remote::new(*remote, refs))
}
fn references(&self, remote: &RemoteId) -> Result<Refs, Error> {
fn references_of(&self, remote: &RemoteId) -> Result<Refs, Error> {
// TODO: Only return known refs, eg. heads/ rad/ tags/ etc..
let entries = self
.backend
@ -670,7 +721,7 @@ impl WriteRepository for Repository {
fn sign_refs<G: Signer>(&self, signer: &G) -> Result<SignedRefs<Verified>, Error> {
let remote = signer.public_key();
let refs = self.references(remote)?;
let refs = self.references_of(remote)?;
let signed = refs.signed(signer)?;
signed.save(remote, self)?;
@ -936,7 +987,7 @@ mod tests {
let signed = project.sign_refs(&signer).unwrap();
let remote = project.remote(&alice).unwrap();
let mut unsigned = project.references(&alice).unwrap();
let mut unsigned = project.references_of(&alice).unwrap();
// The signed refs doesn't contain the signature ref itself.
let sigref = (*SIGREFS_BRANCH).to_ref_string();

2
radicle/src/test/storage.rs
View File

@ -116,7 +116,7 @@ impl ReadRepository for MockRepository {
todo!()
}
fn references(&self, _remote: &RemoteId) -> Result<crate::storage::refs::Refs, Error> {
fn references_of(&self, _remote: &RemoteId) -> Result<crate::storage::refs::Refs, Error> {
todo!()
}