From 2a1154ef87fb35ef62550345762d7ee0b6dfebdb Mon Sep 17 00:00:00 2001 From: Alexis Sellier Date: Tue, 25 Oct 2022 16:46:04 +0200 Subject: [PATCH] Re-work git transports The changes in this commit are all intertwined and it was hard to split them into smaller commits: * Use custom transports everywhere Instead of `file://` and instead of using a child process for fetching and pushing; we go through our custom transports, which use the `rad://` and `heartwood://` schemes. * Introduce a 'mock' remote transport To test the remote transport without needing to spawn TCP streams, we add a mock transport that works between storages on the file system. * Get rid of node's `git-url` Instead of git urls, nodes simply use their node-ids to replicate. --- radicle-node/src/main.rs | 16 +- radicle-node/src/service.rs | 65 ++++---- radicle-node/src/service/config.rs | 9 -- radicle-node/src/service/message.rs | 5 +- radicle-node/src/service/peer.rs | 1 - radicle-node/src/test/peer.rs | 38 ++--- radicle-node/src/test/tests.rs | 35 +---- radicle-node/src/wire.rs | 22 --- radicle-node/src/wire/message.rs | 17 +- radicle/src/git.rs | 44 ++---- radicle/src/profile.rs | 4 + radicle/src/rad.rs | 68 ++++---- radicle/src/storage.rs | 33 ++-- radicle/src/storage/git.rs | 146 ++++++++---------- radicle/src/storage/git/transport.rs | 27 ++++ radicle/src/storage/git/transport/local.rs | 110 +++++++++++++ .../src/storage/git/transport/local/url.rs | 17 +- radicle/src/storage/git/transport/remote.rs | 99 ++++-------- .../src/storage/git/transport/remote/mock.rs | 85 ++++++++++ .../src/storage/git/transport/remote/url.rs | 8 +- radicle/src/test/fixtures.rs | 11 +- radicle/src/test/storage.rs | 19 +-- 22 files changed, 483 insertions(+), 396 deletions(-) create mode 100644 radicle/src/storage/git/transport/remote/mock.rs diff --git a/radicle-node/src/main.rs b/radicle-node/src/main.rs index 838941e0..e09f0ea0 100644 --- a/radicle-node/src/main.rs +++ b/radicle-node/src/main.rs @@ -5,7 +5,7 @@ use anyhow::Context as _; use radicle_node::logger; use radicle_node::prelude::Address; -use radicle_node::{client, control, git, service}; +use radicle_node::{client, control, service}; type Reactor = nakamoto_net_poll::Reactor; @@ -13,7 +13,6 @@ type Reactor = nakamoto_net_poll::Reactor; struct Options { connect: Vec
, listen: Vec, - git_url: git::Url, } impl Options { @@ -22,7 +21,6 @@ impl Options { let mut parser = lexopt::Parser::from_env(); let mut connect = Vec::new(); let mut listen = Vec::new(); - let mut git_url = None; while let Some(arg) = parser.next()? { match arg { @@ -34,11 +32,6 @@ impl Options { let addr = parser.value()?.parse()?; listen.push(addr); } - Long("git-url") => { - let url = git::Url::from_bytes(parser.value()?.into_string()?.as_bytes()) - .map_err(|e| format!("invalid URL: {}", e))?; - git_url = Some(url); - } Long("help") => { println!("usage: radicle-node [--connect ].."); process::exit(0); @@ -46,11 +39,7 @@ impl Options { _ => return Err(arg.unexpected()), } } - Ok(Self { - connect, - listen, - git_url: git_url.ok_or("a Git URL must be specified with `--git-url`")?, - }) + Ok(Self { connect, listen }) } } @@ -65,7 +54,6 @@ fn main() -> anyhow::Result<()> { let config = client::Config { service: service::Config { connect: options.connect, - git_url: options.git_url, ..service::Config::default() }, listen: options.listen, diff --git a/radicle-node/src/service.rs b/radicle-node/src/service.rs index 35a26103..293829bb 100644 --- a/radicle-node/src/service.rs +++ b/radicle-node/src/service.rs @@ -19,7 +19,7 @@ use nakamoto_net as nakamoto; use nakamoto_net::Link; use nonempty::NonEmpty; use radicle::node::Features; -use radicle::storage::ReadStorage; +use radicle::storage::{Namespaces, ReadStorage}; use crate::address; use crate::address::AddressBook; @@ -27,7 +27,6 @@ use crate::clock::{RefClock, Timestamp}; use crate::crypto; use crate::crypto::{Signer, Verified}; use crate::git; -use crate::git::Url; use crate::identity::{Doc, Id}; use crate::node; use crate::service::config::ProjectTracking; @@ -75,7 +74,7 @@ pub type NodeId = crypto::PublicKey; #[derive(Debug, Clone)] pub enum Event { RefsFetched { - from: Url, + from: NodeId, project: Id, updated: Vec, }, @@ -447,15 +446,8 @@ where .ok(); // TODO: Limit the number of seeds we fetch from? Randomize? - for (_, peer) in seeds { - match repo.fetch(&Url { - scheme: git::url::Scheme::Git, - host: Some(peer.addr.ip().to_string()), - port: Some(peer.addr.port()), - // TODO: Fix upstream crate so that it adds a `/` when needed. - path: format!("/{}", id).into(), - ..Url::default() - }) { + for (peer_id, peer) in seeds { + match repo.fetch(&peer_id, Namespaces::default()) { Ok(updated) => { results_ .send(FetchResult::Fetched { @@ -609,7 +601,6 @@ where pub fn handle_announcement( &mut self, session: &NodeId, - git: &git::Url, announcement: &Announcement, ) -> Result { if !announcement.verify() { @@ -635,7 +626,7 @@ where return Ok(false); } - if let Err(err) = self.process_inventory(&message.inventory, *node, git) { + if let Err(err) = self.process_inventory(&message.inventory, *node) { error!("Error processing inventory from {}: {}", node, err); if let Error::Fetch(storage::FetchError::Verify(err)) = err { @@ -665,11 +656,25 @@ where // Refs are only supposed to be relayed by peers who are tracking // the resource. Therefore, it's safe to fetch from the remote // peer, even though it isn't the announcer. - let updated = self.storage.fetch(message.id, git).unwrap(); + let updated = match self + .storage + .repository(message.id) + .map_err(storage::FetchError::from) + .and_then(|mut r| r.fetch(session, Namespaces::default())) + { + Ok(updated) => updated, + Err(err) => { + error!( + "Error fetching repository {} from {}: {}", + message.id, session, err + ); + return Ok(false); + } + }; let is_updated = !updated.is_empty(); self.reactor.event(Event::RefsFetched { - from: git.clone(), + from: *session, project: message.id, updated, }); @@ -748,15 +753,7 @@ where debug!("Received {:?} from {}", &message, peer.ip()); match (&mut peer.state, message) { - ( - SessionState::Initial, - Message::Initialize { - id, - version, - addrs, - git, - }, - ) => { + (SessionState::Initial, Message::Initialize { id, version, addrs }) => { if version != PROTOCOL_VERSION { return Err(SessionError::WrongVersion(version)); } @@ -780,7 +777,6 @@ where id, since: self.clock.local_time(), addrs, - git, ping: Default::default(), }; } @@ -792,12 +788,11 @@ where return Err(SessionError::Misbehavior); } // Process a peer announcement. - (SessionState::Negotiated { id, git, .. }, Message::Announcement(ann)) => { - let git = git.clone(); + (SessionState::Negotiated { id, .. }, Message::Announcement(ann)) => { let id = *id; // Returning true here means that the message should be relayed. - if self.handle_announcement(&id, &git, &ann)? { + if self.handle_announcement(&id, &ann)? { self.gossip.received(ann.clone(), ann.message.timestamp()); // Choose peers we should relay this message to. @@ -857,12 +852,7 @@ where } /// Process a peer inventory announcement by updating our routing table. - fn process_inventory( - &mut self, - inventory: &Inventory, - from: NodeId, - remote: &Url, - ) -> Result<(), Error> { + fn process_inventory(&mut self, inventory: &Inventory, from: NodeId) -> Result<(), Error> { for proj_id in inventory { // TODO: Fire an event on routing update. // FIXME: The timestamp we insert should be the announcement timestamp. @@ -871,7 +861,7 @@ where .insert(*proj_id, from, self.clock.timestamp())? && self.config.is_tracking(proj_id) { - self.storage.fetch(*proj_id, remote)?; + log::info!("Routing table updated for {} with seed {}", proj_id, from); } } Ok(()) @@ -1198,7 +1188,6 @@ mod gossip { signer: &G, config: &Config, ) -> [Message; 4] { - let git = config.git_url.clone(); let inventory = match storage.inventory() { Ok(i) => i, Err(e) => { @@ -1210,7 +1199,7 @@ mod gossip { }; [ - Message::init(*signer.public_key(), config.listen.clone(), git), + Message::init(*signer.public_key(), config.listen.clone()), Message::node(gossip::node(timestamp, config), signer), Message::inventory(gossip::inventory(timestamp, inventory), signer), Message::subscribe(config.filter(), timestamp, Timestamp::MAX), diff --git a/radicle-node/src/service/config.rs b/radicle-node/src/service/config.rs index 455fb352..8017d35e 100644 --- a/radicle-node/src/service/config.rs +++ b/radicle-node/src/service/config.rs @@ -1,6 +1,4 @@ use crate::collections::HashSet; -use crate::git; -use crate::git::Url; use crate::identity::{Id, PublicKey}; use crate::service::filter::Filter; use crate::service::message::Address; @@ -58,8 +56,6 @@ pub struct Config { pub relay: bool, /// List of addresses to listen on for protocol connections. pub listen: Vec
, - /// Our Git URL for fetching projects. - pub git_url: Url, } impl Default for Config { @@ -71,11 +67,6 @@ impl Default for Config { remote_tracking: RemoteTracking::default(), relay: true, listen: vec![], - git_url: Url { - scheme: git::url::Scheme::File, - path: "/dev/null".to_owned().into(), - ..Url::default() - }, } } } diff --git a/radicle-node/src/service/message.rs b/radicle-node/src/service/message.rs index 2fff171e..7d029601 100644 --- a/radicle-node/src/service/message.rs +++ b/radicle-node/src/service/message.rs @@ -4,7 +4,6 @@ use std::{fmt, io, mem, net}; use thiserror::Error; use crate::crypto; -use crate::git; use crate::identity::Id; use crate::node; use crate::service::filter::Filter; @@ -297,7 +296,6 @@ pub enum Message { id: NodeId, version: u32, addrs: Vec
, - git: git::Url, }, /// Subscribe to gossip messages matching the filter and time range. @@ -321,11 +319,10 @@ pub enum Message { } impl Message { - pub fn init(id: NodeId, addrs: Vec
, git: git::Url) -> Self { + pub fn init(id: NodeId, addrs: Vec
) -> Self { Self::Initialize { id, version: PROTOCOL_VERSION, - git, addrs, } } diff --git a/radicle-node/src/service/peer.rs b/radicle-node/src/service/peer.rs index d7122e6a..0f9481e5 100644 --- a/radicle-node/src/service/peer.rs +++ b/radicle-node/src/service/peer.rs @@ -28,7 +28,6 @@ pub enum SessionState { since: LocalTime, /// Addresses this peer is reachable on. addrs: Vec
, - git: Url, ping: PingState, }, /// When a peer is disconnected. diff --git a/radicle-node/src/test/peer.rs b/radicle-node/src/test/peer.rs index ccb2c8f5..b575892c 100644 --- a/radicle-node/src/test/peer.rs +++ b/radicle-node/src/test/peer.rs @@ -9,8 +9,6 @@ use log::*; use crate::address; use crate::clock::{RefClock, Timestamp}; use crate::crypto::Signer; -use crate::git; -use crate::git::Url; use crate::identity::Id; use crate::node; use crate::prelude::NodeId; @@ -19,7 +17,8 @@ use crate::service::config::*; use crate::service::message::*; use crate::service::reactor::Io; use crate::service::*; -use crate::storage::WriteStorage; +use crate::storage::git::transport::remote; +use crate::storage::{RemoteId, WriteStorage}; use crate::test::arbitrary; use crate::test::signer::MockSigner; use crate::test::simulator; @@ -73,26 +72,10 @@ where S: WriteStorage + 'static, { pub fn new(name: &'static str, ip: impl Into, storage: S) -> Self { - let git_url = Url { - scheme: git::url::Scheme::File, - path: storage.path().to_string_lossy().to_string().into(), - - ..git::Url::default() - }; let mut rng = fastrand::Rng::new(); let signer = MockSigner::new(&mut rng); - Self::config( - name, - Config { - git_url, - ..Config::default() - }, - ip, - storage, - signer, - rng, - ) + Self::config(name, Config::default(), ip, storage, signer, rng) } } @@ -145,8 +128,12 @@ where self.service.clock().timestamp() } - pub fn git_url(&self) -> Url { - self.config().git_url.clone() + pub fn git_url(&self, repo: Id, namespace: Option) -> remote::Url { + remote::Url { + node: self.node_id(), + repo, + namespace, + } } pub fn node_id(&self) -> NodeId { @@ -197,14 +184,12 @@ where pub fn connect_from(&mut self, peer: &Self) { let remote = simulator::Peer::::addr(peer); let local = net::SocketAddr::new(self.ip, self.rng.u16(..)); - let git = format!("file:///{}.git", remote.ip()); - let git = Url::from_bytes(git.as_bytes()).unwrap(); self.initialize(); self.service.connected(remote, &local, Link::Inbound); self.receive( &remote, - Message::init(peer.node_id(), vec![Address::from(remote)], git), + Message::init(peer.node_id(), vec![Address::from(remote)]), ); let mut msgs = self.messages(&remote); @@ -244,10 +229,9 @@ where }) .expect("`inventory-announcement` is sent"); - let git = peer.config().git_url.clone(); self.receive( &remote, - Message::init(peer.node_id(), peer.config().listen.clone(), git), + Message::init(peer.node_id(), peer.config().listen.clone()), ); } diff --git a/radicle-node/src/test/tests.rs b/radicle-node/src/test/tests.rs index 06910b94..a82fe14b 100644 --- a/radicle-node/src/test/tests.rs +++ b/radicle-node/src/test/tests.rs @@ -13,6 +13,7 @@ use crate::service::peer::*; use crate::service::reactor::Io; use crate::service::ServiceState as _; use crate::service::*; +use crate::storage::git::transport::{local, remote}; use crate::storage::git::Storage; use crate::storage::ReadStorage; use crate::test::arbitrary; @@ -223,16 +224,6 @@ fn test_inventory_sync() { let seeds = alice.routing().get(proj).unwrap(); assert!(seeds.contains(&bob.node_id())); } - - let a = alice - .storage() - .inventory() - .unwrap() - .into_iter() - .collect::>(); - let b = projs.into_iter().collect::>(); - - assert_eq!(a, b); } #[test] @@ -441,22 +432,7 @@ fn test_refs_announcement_relay() { let signer = MockSigner::new(&mut rng); let storage = fixtures::storage(tmp.path().join("bob"), &signer).unwrap(); - Peer::config( - "bob", - Config { - git_url: git::Url { - scheme: git::url::Scheme::File, - path: storage.path().to_string_lossy().to_string().into(), - - ..git::Url::default() - }, - ..Config::default() - }, - [9, 9, 9, 9], - storage, - signer, - rng, - ) + Peer::config("bob", Config::default(), [9, 9, 9, 9], storage, signer, rng) }; let bob_inv = bob.inventory().unwrap(); @@ -687,6 +663,11 @@ fn test_push_and_pull() { let storage_eve = Storage::open(tempdir.path().join("eve").join("storage")).unwrap(); let mut eve = Peer::new("eve", [9, 9, 9, 9], storage_eve); + remote::mock::register(&alice.node_id(), alice.storage().path()); + remote::mock::register(&eve.node_id(), eve.storage().path()); + remote::mock::register(&bob.node_id(), bob.storage().path()); + local::register(alice.storage().clone()); + // Alice and Bob connect to Eve. alice.command(service::Command::Connect(eve.addr())); bob.command(service::Command::Connect(eve.addr())); @@ -744,7 +725,7 @@ fn test_push_and_pull() { assert_matches!( sim.events(&bob.ip).next(), Some(service::Event::RefsFetched { from, .. }) - if from == eve.git_url(), + if from == eve.node_id(), "Bob fetched from Eve" ); } diff --git a/radicle-node/src/wire.rs b/radicle-node/src/wire.rs index 84711d79..d2014031 100644 --- a/radicle-node/src/wire.rs +++ b/radicle-node/src/wire.rs @@ -348,16 +348,6 @@ impl Decode for String { } } -impl Decode for git::Url { - fn decode(reader: &mut R) -> Result { - let url = String::decode(reader)?; - let url = Self::from_bytes(url.as_bytes()) - .map_err(|error| Error::InvalidGitUrl { url, error })?; - - Ok(url) - } -} - impl Decode for Id { fn decode(reader: &mut R) -> Result { let oid: git::Oid = Decode::decode(reader)?; @@ -648,18 +638,6 @@ mod tests { ); } - #[test] - fn test_git_url() { - let url = git::Url { - scheme: git::url::Scheme::Https, - path: "/git".to_owned().into(), - host: Some("seed.radicle.xyz".to_owned()), - port: Some(8888), - ..git::Url::default() - }; - assert_eq!(deserialize::(&serialize(&url)).unwrap(), url); - } - #[test] fn test_filter_invalid() { let b = bloomy::BloomFilter::with_size(filter::FILTER_SIZE_M / 3); diff --git a/radicle-node/src/wire/message.rs b/radicle-node/src/wire/message.rs index 503653c1..b9ff2ecc 100644 --- a/radicle-node/src/wire/message.rs +++ b/radicle-node/src/wire/message.rs @@ -2,7 +2,6 @@ use std::{io, mem, net}; use byteorder::{NetworkEndian, ReadBytesExt}; -use crate::git; use crate::prelude::*; use crate::service::message::*; use crate::wire; @@ -169,16 +168,10 @@ impl wire::Encode for Message { let mut n = self.type_id().encode(writer)?; match self { - Self::Initialize { - id, - version, - addrs, - git, - } => { + Self::Initialize { id, version, addrs } => { n += id.encode(writer)?; n += version.encode(writer)?; n += addrs.as_slice().encode(writer)?; - n += git.encode(writer)?; } Self::Subscribe(Subscribe { filter, @@ -226,14 +219,8 @@ impl wire::Decode for Message { let id = NodeId::decode(reader)?; let version = u32::decode(reader)?; let addrs = Vec::
::decode(reader)?; - let git = git::Url::decode(reader)?; - Ok(Self::Initialize { - id, - version, - addrs, - git, - }) + Ok(Self::Initialize { id, version, addrs }) } Ok(MessageType::Subscribe) => { let filter = Filter::decode(reader)?; diff --git a/radicle/src/git.rs b/radicle/src/git.rs index 58890170..1521aaaa 100644 --- a/radicle/src/git.rs +++ b/radicle/src/git.rs @@ -22,8 +22,8 @@ pub use git_ref_format::{ lit, name, qualified, refname, Component, Namespaced, Qualified, RefStr, RefString, }; pub use git_url as url; -pub use git_url::Url; pub use radicle_git_ext as ext; +pub use storage::git::transport::local::Url; pub use storage::BranchName; /// Default port of the `git` transport protocol. @@ -265,43 +265,25 @@ pub fn configure_remote<'r>( Ok(remote) } -/// Fetch from the given `remote` using the provided `namespace`. -/// -/// This uses [`Command`] under the hood and is the equivalent to: -/// -/// `GIT_NAMESPACE= git fetch ` -pub fn fetch(repo: &git2::Repository, remote: &str, namespace: &RemoteId) -> io::Result { - run( - &repo.path(), - ["fetch", remote], - [("GIT_NAMESPACE", Component::from(namespace).as_str())], - ) +/// Fetch from the given `remote`. +pub fn fetch(repo: &git2::Repository, remote: &str) -> Result<(), git2::Error> { + repo.find_remote(remote)?.fetch::<&str>(&[], None, None) } /// Push `refspecs` to the given `remote` using the provided `namespace`. -/// -/// This uses [`Command`] under the hood and is the equivalent to: -/// -/// `GIT_NAMESPACE= git push []` -pub fn push( +pub fn push<'a>( repo: &git2::Repository, remote: &str, - namespace: &RemoteId, - refspecs: impl IntoIterator, -) -> io::Result -where - Ref: AsRef, -{ - let mut args = vec!["push".to_owned(), remote.to_owned()]; + refspecs: impl IntoIterator, &'a Qualified<'a>)>, +) -> Result<(), git2::Error> { let refspecs = refspecs .into_iter() - .map(|(src, dst)| format!("{}:{}", src.as_ref().as_str(), dst.as_ref().as_str())); - args.extend(refspecs); - run( - &repo.path(), - args, - [("GIT_NAMESPACE", Component::from(namespace).as_str())], - ) + .map(|(src, dst)| format!("{}:{}", src.as_str(), dst.as_str())); + + repo.find_remote(remote)? + .push(refspecs.collect::>().as_slice(), None)?; + + Ok(()) } /// Set the upstream of the given branch to the given remote. diff --git a/radicle/src/profile.rs b/radicle/src/profile.rs index 55d2f154..b7d87fa6 100644 --- a/radicle/src/profile.rs +++ b/radicle/src/profile.rs @@ -16,6 +16,7 @@ use std::{env, io}; use crate::crypto::{KeyPair, PublicKey, SecretKey, Signature, Signer}; use crate::keystore::{Error, UnsafeKeystore}; use crate::node; +use crate::storage::git::transport; use crate::storage::git::Storage; #[derive(Debug)] @@ -52,6 +53,7 @@ impl Profile { }; let storage = Storage::open(&home.join("storage"))?; + transport::local::register(storage.clone()); keystore.put(&signer.public, &signer.secret)?; Ok(Profile { @@ -67,6 +69,8 @@ impl Profile { let signer = UnsafeSigner { public, secret }; let storage = Storage::open(&home.join("storage"))?; + transport::local::register(storage.clone()); + Ok(Profile { home, signer, diff --git a/radicle/src/rad.rs b/radicle/src/rad.rs index 2bb5447f..02ac3fac 100644 --- a/radicle/src/rad.rs +++ b/radicle/src/rad.rs @@ -11,7 +11,8 @@ use crate::git; use crate::identity::project::DocError; use crate::identity::Id; use crate::node; -use crate::storage::git::ProjectError; +use crate::storage::git::transport::{self, remote}; +use crate::storage::git::{ProjectError, Storage}; use crate::storage::refs::SignedRefs; use crate::storage::{BranchName, ReadRepository as _, RemoteId, WriteRepository as _}; use crate::{identity, storage}; @@ -41,13 +42,13 @@ pub enum InitError { } /// Initialize a new radicle project from a git repository. -pub fn init( +pub fn init( repo: &git2::Repository, name: &str, description: &str, default_branch: BranchName, signer: G, - storage: &S, + storage: &Storage, ) -> Result<(Id, SignedRefs), InitError> { // TODO: Better error when project id already exists in storage, but remote doesn't. let pk = signer.public_key(); @@ -65,7 +66,7 @@ pub fn init( .verified()?; let (id, _, project) = doc.create(pk, "Initialize Radicle\n", storage)?; - let url = storage.url(&id); + let url = git::Url::from(id).with_namespace(*pk); git::set_upstream( repo, @@ -75,7 +76,14 @@ pub fn init( )?; git::configure_remote(repo, &REMOTE_NAME, &url)?; - git::push(repo, &REMOTE_NAME, pk, [(&default_branch, &default_branch)])?; + git::push( + repo, + &REMOTE_NAME, + [( + &git::fmt::lit::refs_heads(&default_branch).into(), + &git::fmt::lit::refs_heads(&default_branch).into(), + )], + )?; let signed = project.sign_refs(signer)?; let _head = project.set_head()?; @@ -205,6 +213,8 @@ pub fn clone, G: Signer, S: storage::WriteStorage, H: node::Handl #[derive(Error, Debug)] pub enum CloneUrlError { + #[error("missing namespace in url")] + MissingNamespace, #[error("storage: {0}")] Storage(#[from] storage::Error), #[error("fetch: {0}")] @@ -216,16 +226,16 @@ pub enum CloneUrlError { } pub fn clone_url, G: Signer, S: storage::WriteStorage>( - proj: Id, - url: &git::Url, + url: &remote::Url, path: P, signer: &G, storage: &S, ) -> Result { - let mut project = storage.repository(proj)?; - let _updates = project.fetch(url)?; - let _ = fork(proj, signer, storage)?; - let working = checkout(proj, signer.public_key(), path, storage)?; + let namespace = url.namespace.ok_or(CloneUrlError::MissingNamespace)?; + let mut project = storage.repository(url.repo)?; + let _updates = project.fetch(&url.node, namespace)?; + let _ = fork(url.repo, signer, storage)?; + let working = checkout(url.repo, signer.public_key(), path, storage)?; Ok(working) } @@ -233,7 +243,7 @@ pub fn clone_url, G: Signer, S: storage::WriteStorage>( #[derive(Error, Debug)] pub enum CheckoutError { #[error("failed to fetch to working copy")] - Fetch(#[source] io::Error), + Fetch(#[source] git2::Error), #[error("git: {0}")] Git(#[from] git2::Error), #[error("storage: {0}")] @@ -262,11 +272,11 @@ pub fn checkout, S: storage::ReadStorage>( opts.no_reinit(true).description(&project.description); let repo = git2::Repository::init_opts(path.as_ref().join(&project.name), &opts)?; - let url = storage.url(&proj); + let url = git::Url::from(proj).with_namespace(*remote); // Configure and fetch all refs from remote. git::configure_remote(&repo, &REMOTE_NAME, &url)?; - git::fetch(&repo, &REMOTE_NAME, remote).map_err(CheckoutError::Fetch)?; + git::fetch(&repo, &REMOTE_NAME).map_err(CheckoutError::Fetch)?; { // Setup default branch. @@ -293,23 +303,18 @@ pub enum RemoteError { #[error("git: {0}")] Git(#[from] git2::Error), #[error("invalid remote url: {0}")] - Url(#[from] git::url::parse::Error), - #[error("remote url doesn't have an id: `{0}`")] - MissingId(git::Url), - #[error("identifier error: {0}")] - InvalidId(#[from] identity::IdError), + Url(#[from] transport::local::UrlError), + #[error("invalid utf-8 string")] + InvalidUtf8, } /// Get the radicle ("rad") remote of a repository, and return the associated project id. pub fn remote(repo: &git2::Repository) -> Result<(git2::Remote<'_>, Id), RemoteError> { let remote = repo.find_remote(&REMOTE_NAME)?; - let url = remote.url_bytes(); - let url = git::Url::from_bytes(url)?; - let path = url.path.to_string(); - let id = path.split('/').last().ok_or(RemoteError::MissingId(url))?; - let id = Id::from_str(id)?; + let url = remote.url().ok_or(RemoteError::InvalidUtf8)?; + let url = git::Url::from_str(url)?; - Ok((remote, id)) + Ok((remote, url.repo)) } #[cfg(test)] @@ -319,6 +324,7 @@ mod tests { use super::*; use crate::git::fmt::refname; use crate::identity::{Delegate, Did}; + use crate::storage::git::transport; use crate::storage::git::Storage; use crate::storage::{ReadStorage, WriteStorage}; use crate::test::{fixtures, signer::MockSigner}; @@ -329,8 +335,10 @@ mod tests { let signer = MockSigner::default(); let public_key = *signer.public_key(); let storage = Storage::open(tempdir.path().join("storage")).unwrap(); - let (repo, _) = fixtures::repository(tempdir.path().join("working")); + transport::local::register(storage.clone()); + + let (repo, _) = fixtures::repository(tempdir.path().join("working")); let (proj, refs) = init( &repo, "acme", @@ -384,9 +392,11 @@ mod tests { let bob = MockSigner::new(&mut rng); let bob_id = bob.public_key(); let storage = Storage::open(tempdir.path().join("storage")).unwrap(); - let (original, _) = fixtures::repository(tempdir.path().join("original")); + + transport::local::register(storage.clone()); // Alice creates a project. + let (original, _) = fixtures::repository(tempdir.path().join("original")); let (id, alice_refs) = init( &original, "acme", @@ -415,8 +425,10 @@ mod tests { let signer = MockSigner::default(); let remote_id = signer.public_key(); let storage = Storage::open(tempdir.path().join("storage")).unwrap(); - let (original, _) = fixtures::repository(tempdir.path().join("original")); + transport::local::register(storage.clone()); + + let (original, _) = fixtures::repository(tempdir.path().join("original")); let (id, _) = init( &original, "acme", diff --git a/radicle/src/storage.rs b/radicle/src/storage.rs index cc54a187..1854040e 100644 --- a/radicle/src/storage.rs +++ b/radicle/src/storage.rs @@ -15,7 +15,6 @@ use crate::collections::HashMap; use crate::crypto; use crate::crypto::{PublicKey, Signer, Unverified, Verified}; use crate::git::ext as git_ext; -use crate::git::Url; use crate::git::{Qualified, RefError, RefString}; use crate::identity; use crate::identity::{Id, IdError}; @@ -26,6 +25,22 @@ use self::refs::SignedRefs; pub type BranchName = git::RefString; pub type Inventory = Vec; +/// Describes one or more namespaces. +#[derive(Default, Debug)] +pub enum Namespaces { + /// All namespaces. + #[default] + All, + /// A single namespace, by public key. + One(PublicKey), +} + +impl From for Namespaces { + fn from(pk: PublicKey) -> Self { + Self::One(pk) + } +} + /// Storage error. #[derive(Error, Debug)] pub enum Error { @@ -217,7 +232,6 @@ impl Remote { pub trait ReadStorage { fn path(&self) -> &Path; - fn url(&self, proj: &Id) -> Url; fn get( &self, remote: &RemoteId, @@ -230,7 +244,6 @@ pub trait WriteStorage: ReadStorage { type Repository: WriteRepository; fn repository(&self, proj: Id) -> Result; - fn fetch(&self, proj_id: Id, remote: &Url) -> Result, FetchError>; } pub trait ReadRepository { @@ -286,7 +299,11 @@ pub trait ReadRepository { } pub trait WriteRepository: ReadRepository { - fn fetch(&mut self, url: &Url) -> Result, FetchError>; + fn fetch( + &mut self, + node: &RemoteId, + namespaces: impl Into, + ) -> Result, FetchError>; fn set_head(&self) -> Result; fn sign_refs(&self, signer: G) -> Result, Error>; fn raw(&self) -> &git2::Repository; @@ -301,10 +318,6 @@ where self.deref().path() } - fn url(&self, proj: &Id) -> Url { - self.deref().url(proj) - } - fn inventory(&self) -> Result { self.deref().inventory() } @@ -328,10 +341,6 @@ where fn repository(&self, proj: Id) -> Result { self.deref().repository(proj) } - - fn fetch(&self, proj_id: Id, remote: &Url) -> Result, FetchError> { - self.deref().fetch(proj_id, remote) - } } #[cfg(test)] diff --git a/radicle/src/storage/git.rs b/radicle/src/storage/git.rs index ee1cbb87..53e5a701 100644 --- a/radicle/src/storage/git.rs +++ b/radicle/src/storage/git.rs @@ -2,7 +2,7 @@ pub mod transport; use std::collections::{BTreeMap, HashMap}; use std::path::{Path, PathBuf}; -use std::{fmt, fs, io}; +use std::{fs, io}; use git_ref_format::refspec; use once_cell::sync::Lazy; @@ -21,7 +21,8 @@ use crate::storage::{ pub use crate::git::*; -use super::{RefUpdate, RemoteId}; +use super::{Namespaces, RefUpdate, RemoteId}; +use transport::remote; pub static REMOTES_GLOB: Lazy = Lazy::new(|| refspec::pattern!("refs/namespaces/*")); @@ -59,32 +60,16 @@ impl ProjectError { } } +#[derive(Debug, Clone)] pub struct Storage { path: PathBuf, } -impl fmt::Debug for Storage { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - write!(f, "Storage(..)") - } -} - impl ReadStorage for Storage { fn path(&self) -> &Path { self.path.as_path() } - fn url(&self, proj: &Id) -> Url { - let path = paths::repository(self, proj); - - Url { - scheme: git_url::Scheme::File, - path: path.to_string_lossy().to_string().into(), - - ..git::Url::default() - } - } - fn get(&self, remote: &RemoteId, proj: Id) -> Result>, ProjectError> { // TODO: Don't create a repo here if it doesn't exist? // Perhaps for checking we could have a `contains` method? @@ -107,19 +92,6 @@ impl WriteStorage for Storage { fn repository(&self, proj: Id) -> Result { Repository::open(paths::repository(self, &proj), proj) } - - fn fetch(&self, proj_id: Id, remote: &Url) -> Result, FetchError> { - let mut repo = self.repository(proj_id)?; - let mut path = remote.path.clone(); - - path.push(b'/'); - path.extend(proj_id.to_string().into_bytes()); - - repo.fetch(&Url { - path, - ..remote.clone() - }) - } } impl Storage { @@ -200,7 +172,7 @@ impl Repository { let backend = match git2::Repository::open_bare(path.as_ref()) { Err(e) if ext::is_not_found_err(&e) => { let backend = git2::Repository::init_opts( - path, + &path, git2::RepositoryInitOptions::new() .bare(true) .no_reinit(true) @@ -546,9 +518,11 @@ impl WriteRepository for Repository { /// with pruning *on*, and discard the staging copy. If it fails, we just discard the /// staging copy. /// - fn fetch(&mut self, url: &git::Url) -> Result, FetchError> { - // TODO: Have function to fetch specific remotes. - // + fn fetch( + &mut self, + node: &RemoteId, + namespaces: impl Into, + ) -> Result, FetchError> { // The steps are summarized in the following diagram: // // staging <- git-clone -- local (canonical) # create staging copy @@ -558,8 +532,12 @@ impl WriteRepository for Repository { // // local <- git-fetch -- staging # fetch from staging copy // - let url = url.to_string(); - let refs: &[&str] = &["refs/namespaces/*:refs/namespaces/*"]; + + let namespace = match namespaces.into() { + Namespaces::All => None, + Namespaces::One(ns) => Some(ns), + }; + let mut updates = Vec::new(); let mut callbacks = git2::RemoteCallbacks::new(); let tempdir = tempfile::tempdir()?; @@ -574,10 +552,10 @@ impl WriteRepository for Repository { // TODO: Due to this, I think we'll have to run GC when there is a failure. .clone_local(git2::build::CloneLocal::Local) .clone( - &git::Url { + &git::url::Url { scheme: git::url::Scheme::File, path: self.backend.path().to_string_lossy().to_string().into(), - ..git::Url::default() + ..git::url::Url::default() } .to_string(), &path, @@ -589,8 +567,16 @@ impl WriteRepository for Repository { // Fetch from the remote into the staging copy. staging_repo - .remote_anonymous(&url)? - .fetch(refs, Some(&mut opts), None)?; + .remote_anonymous( + remote::Url { + node: *node, + repo: self.id, + namespace, + } + .to_string() + .as_str(), + )? + .fetch(&["refs/*:refs/*"], Some(&mut opts), None)?; // Verify the staging copy as if it was the canonical copy. Repository { @@ -617,21 +603,26 @@ impl WriteRepository for Repository { { let mut remote = self.backend.remote_anonymous( - &git::Url { + &git::url::Url { scheme: git::url::Scheme::File, path: staging.to_string_lossy().to_string().into(), - ..git::Url::default() + ..git::url::Url::default() } .to_string(), )?; let mut opts = git2::FetchOptions::default(); opts.remote_callbacks(callbacks); + let refspec = if let Some(namespace) = namespace { + format!("refs/namespaces/{namespace}/refs/*:refs/namespaces/{namespace}/refs/*") + } else { + "refs/namespaces/*:refs/namespaces/*".to_owned() + }; // TODO: Make sure we verify before pruning, as pruning may get us into // a state we can't roll back. opts.prune(git2::FetchPrune::On); // Fetch from the staging copy into the canonical repo. - remote.fetch(refs, Some(&mut opts), None)?; + remote.fetch(&[refspec], Some(&mut opts), None)?; } // Set repository HEAD for git cloning support. self.set_head()?; @@ -742,15 +733,7 @@ mod tests { let storage = fixtures::storage(dir.path(), &signer).unwrap(); let inv = storage.inventory().unwrap(); let proj = inv.first().unwrap(); - let mut refs = git::remote_refs(&git::Url { - scheme: git_url::Scheme::File, - path: paths::repository(&storage, proj) - .to_string_lossy() - .into_owned() - .into(), - ..git::Url::default() - }) - .unwrap(); + let mut refs = git::remote_refs(&git::Url::from(*proj)).unwrap(); let project = storage.repository(*proj).unwrap(); let remotes = project.remotes().unwrap(); @@ -773,6 +756,7 @@ mod tests { fn test_fetch() { let tmp = tempfile::tempdir().unwrap(); let alice_signer = MockSigner::default(); + let alice_pk = *alice_signer.public_key(); let alice = fixtures::storage(tmp.path().join("alice"), alice_signer).unwrap(); let bob = Storage::open(tmp.path().join("bob")).unwrap(); let inventory = alice.inventory().unwrap(); @@ -785,17 +769,10 @@ mod tests { let updates = bob .repository(proj) .unwrap() - .fetch(&git::Url { - scheme: git_url::Scheme::File, - path: paths::repository(&alice, &proj) - .to_string_lossy() - .into_owned() - .into(), - ..git::Url::default() - }) + .fetch(&alice_pk, alice_pk) .unwrap(); - // Four refs are created for each remote. + // Three refs are created for each remote. assert_eq!(updates.len(), remotes.len() * 3); for update in updates { @@ -831,24 +808,20 @@ mod tests { let tmp = tempfile::tempdir().unwrap(); let alice = Storage::open(tmp.path().join("alice/storage")).unwrap(); let bob = Storage::open(tmp.path().join("bob/storage")).unwrap(); - let alice_signer = MockSigner::new(&mut fastrand::Rng::new()); let alice_id = alice_signer.public_key(); let (proj_id, _, proj_repo, alice_head) = fixtures::project(tmp.path().join("alice/project"), &alice, &alice_signer).unwrap(); - let refname = Qualified::from_refstr(git::refname!("refs/heads/master")).unwrap(); - let alice_url = git::Url { - scheme: git_url::Scheme::File, - path: paths::repository(&alice, &proj_id) - .to_string_lossy() - .into_owned() - .into(), - ..git::Url::default() - }; + + transport::remote::mock::register(alice_id, alice.path()); // Have Bob fetch Alice's refs. - let updates = bob.repository(proj_id).unwrap().fetch(&alice_url).unwrap(); + let updates = bob + .repository(proj_id) + .unwrap() + .fetch(alice_signer.public_key(), *alice_signer.public_key()) + .unwrap(); // Three refs are created: the branch, the signature and the id. assert_eq!(updates.len(), 3); @@ -857,12 +830,16 @@ mod tests { let alice_head = git::commit(&proj_repo, &alice_head, &refname, "Making changes", "Alice") .unwrap() .id(); - git::push(&proj_repo, "rad", alice_id, [(&refname, &refname)]).unwrap(); + git::push(&proj_repo, "rad", [(&refname, &refname)]).unwrap(); alice_proj_storage.sign_refs(&alice_signer).unwrap(); alice_proj_storage.set_head().unwrap(); // Have Bob fetch Alice's new commit. - let updates = bob.repository(proj_id).unwrap().fetch(&alice_url).unwrap(); + let updates = bob + .repository(proj_id) + .unwrap() + .fetch(alice_signer.public_key(), *alice_signer.public_key()) + .unwrap(); // The branch and signature refs are updated. assert_matches!( updates.as_slice(), @@ -881,6 +858,9 @@ mod tests { let tmp = tempfile::tempdir().unwrap(); let signer = MockSigner::default(); let storage = Storage::open(tmp.path().join("storage")).unwrap(); + + transport::local::register(storage.clone()); + let (id, _, _, _) = fixtures::project(tmp.path().join("project"), &storage, &signer).unwrap(); let proj = storage.repository(id).unwrap(); @@ -897,6 +877,10 @@ mod tests { } #[test] + #[ignore] + // Test the remote transport using `git-upload-pack` and TCP streams. + // Must be run on its own, since it tries to register the remote transport, which + // will fail if the mock transport was already registered. fn test_upload_pack() { let tmp = tempfile::tempdir().unwrap(); let signer = MockSigner::default(); @@ -907,6 +891,9 @@ mod tests { let source_path = tmp.path().join("source"); let target_path = tmp.path().join("target"); let (source, _) = fixtures::repository(&source_path); + + transport::local::register(storage.clone()); + let (proj, _) = rad::init( &source, "radicle", @@ -966,14 +953,11 @@ mod tests { }); opts.remote_callbacks(callbacks); - // Register the `heartwood://` transport. - transport::remote::register().unwrap(); - let target = git2::Repository::init_bare(target_path).unwrap(); let stream = net::TcpStream::connect(addr).unwrap(); - let smart = transport::remote::Smart::singleton(); - smart.insert(proj, Box::new(stream.try_clone().unwrap())); + // Register the `heartwood://` transport for this stream. + transport::remote::register(remote, stream.try_clone().unwrap()); // Fetch with the `heartwood://` transport. target diff --git a/radicle/src/storage/git/transport.rs b/radicle/src/storage/git/transport.rs index 1fc1939a..947054be 100644 --- a/radicle/src/storage/git/transport.rs +++ b/radicle/src/storage/git/transport.rs @@ -1,2 +1,29 @@ pub mod local; pub mod remote; + +use std::{io, process}; + +/// A wrapper around a child process' stdin and stdout, +/// making it [`io::Read`] and [`io::Write`]. +/// +/// Used for some of the git transports. +pub(crate) struct ChildStream { + pub stdin: process::ChildStdin, + pub stdout: process::ChildStdout, +} + +impl io::Read for ChildStream { + fn read(&mut self, buf: &mut [u8]) -> io::Result { + self.stdout.read(buf) + } +} + +impl io::Write for ChildStream { + fn write(&mut self, buf: &[u8]) -> io::Result { + self.stdin.write(buf) + } + + fn flush(&mut self) -> io::Result<()> { + self.stdin.flush() + } +} diff --git a/radicle/src/storage/git/transport/local.rs b/radicle/src/storage/git/transport/local.rs index e8e9ab2b..51a969d3 100644 --- a/radicle/src/storage/git/transport/local.rs +++ b/radicle/src/storage/git/transport/local.rs @@ -1,2 +1,112 @@ +//! The local git transport protocol. pub mod url; pub use url::{Url, UrlError}; + +use std::cell::RefCell; +use std::process; +use std::str::FromStr; +use std::sync::Once; + +use once_cell::sync::OnceCell; + +use crate::storage; +use crate::storage::git::Storage; + +use super::ChildStream; + +thread_local! { + /// Stores a storage instance per thread. + /// This avoids race conditions when used in a multi-threaded context. + static THREAD_STORAGE: OnceCell = OnceCell::default(); +} + +/// Local git transport over the filesystem. +#[derive(Default)] +struct Local { + /// The child process we spawn. + child: RefCell>, +} + +impl git2::transport::SmartSubtransport for Local { + fn action( + &self, + url: &str, + service: git2::transport::Service, + ) -> Result, git2::Error> { + let url = Url::from_str(url).map_err(|e| git2::Error::from_str(e.to_string().as_str()))?; + let service: &str = match service { + git2::transport::Service::UploadPack | git2::transport::Service::UploadPackLs => { + "upload-pack" + } + git2::transport::Service::ReceivePack | git2::transport::Service::ReceivePackLs => { + "receive-pack" + } + }; + let git_dir = THREAD_STORAGE + .with(|t| { + t.get() + .map(|s| storage::git::paths::repository(&s, &url.repo)) + }) + .ok_or_else(|| git2::Error::from_str("local transport storage was not registered"))?; + + let mut cmd = process::Command::new("git"); + + if let Some(ns) = url.namespace { + cmd.env("GIT_NAMESPACE", ns.to_string()); + } + + let mut child = cmd + .arg(service) + .arg(&git_dir) + .stdin(process::Stdio::piped()) + .stdout(process::Stdio::piped()) + .stderr(process::Stdio::inherit()) + .spawn() + .map_err(|e| git2::Error::from_str(e.to_string().as_str()))?; + + let stdin = child.stdin.take().expect("taking stdin is safe"); + let stdout = child.stdout.take().expect("taking stdout is safe"); + + self.child.replace(Some(child)); + + Ok(Box::new(ChildStream { stdout, stdin })) + } + + fn close(&self) -> Result<(), git2::Error> { + if let Some(mut child) = self.child.take() { + let result = child + .wait() + .map_err(|e| git2::Error::from_str(e.to_string().as_str()))?; + + if !result.success() { + return if let Some(code) = result.code() { + Err(git2::Error::from_str( + format!("transport: child process exited with error code {code}").as_str(), + )) + } else { + Err(git2::Error::from_str( + "transport: child process exited with unknown error", + )) + }; + } + } + Ok(()) + } +} + +// TODO: Instead of taking a storage here, we should take something that can return a storage path. +/// Register a storage with the local transport protocol. +pub fn register(storage: Storage) { + static REGISTER: Once = Once::new(); + + THREAD_STORAGE.with(|s| { + s.set(storage).ok(); + }); + + REGISTER.call_once(|| unsafe { + git2::transport::register(Url::SCHEME, move |remote| { + git2::transport::Transport::smart(remote, false, Local::default()) + }) + .expect("local transport registration"); + }); +} diff --git a/radicle/src/storage/git/transport/local/url.rs b/radicle/src/storage/git/transport/local/url.rs index 9efb1652..4f7a107a 100644 --- a/radicle/src/storage/git/transport/local/url.rs +++ b/radicle/src/storage/git/transport/local/url.rs @@ -35,7 +35,7 @@ pub enum UrlError { /// /// `rad://[/]` /// -#[derive(Debug)] +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)] pub struct Url { /// Repository identifier. pub repo: Id, @@ -46,6 +46,21 @@ pub struct Url { impl Url { /// URL scheme. pub const SCHEME: &str = "rad"; + + /// Return this URL with the given namespace added. + pub fn with_namespace(mut self, namespace: Namespace) -> Self { + self.namespace = Some(namespace); + self + } +} + +impl From for Url { + fn from(repo: Id) -> Self { + Self { + repo, + namespace: None, + } + } } impl fmt::Display for Url { diff --git a/radicle/src/storage/git/transport/remote.rs b/radicle/src/storage/git/transport/remote.rs index 08ab71b4..7a9c7a34 100644 --- a/radicle/src/storage/git/transport/remote.rs +++ b/radicle/src/storage/git/transport/remote.rs @@ -3,72 +3,46 @@ //! To have control over the communication, and to allow git streams to be multiplexed over //! existing TCP connections, we implement the [`git2::transport::SmartSubtransport`] trait. //! -//! We choose `rad` as the URL scheme for this custom transport, and include only the identity -//! of the repository we're looking to fetch, eg. `rad://zP1GztjSdYNHK7jpdrXbaJ6Ki2Ke`, since -//! we expect a connection to a host to already be established. +//! We choose `heartwood` as the URL scheme for this custom transport, and include the node we'd +//! like to fetch from, as well as the repository. We expect the TCP stream to already be +//! established when this transport is called. //! -//! We then maintain a map from identifier to stream, for all active streams, ie. streams that -//! are associated with an underlying TCP connection. When a URL is requested, we lookup -//! the stream and return it to the [`git2`] smart-protocol implementation, so that it can carry -//! out the git smart protocol. +//! We then maintain a map from node identifier to stream, for all active TCP connections. When a +//! URL is requested, we lookup the associated stream and return it to the [`git2`] smart-protocol +//! implementation, so that it can carry out the git smart protocol. //! -//! This module is meant to be used by first registering our transport with [`register`] and then -//! adding or removing streams through [`Smart`], which can be obtained via [`Smart::singleton`]. +//! This module is meant to be used by registering streams with [`register`]. +pub mod mock; pub mod url; use std::collections::HashMap; use std::str::FromStr; -use std::sync::atomic; -use std::sync::{Arc, Mutex}; +use std::sync::Mutex; +use std::sync::Once; use git2::transport::SmartSubtransportStream; use once_cell::sync::Lazy; -use crate::identity::Id; +use crate::storage::RemoteId; pub use url::{Url, UrlError}; /// The map of git smart sub-transport streams. We keep a global map because we have /// no control over how [`git2::transport::register`] instantiates our [`Smart`] transport /// or its underlying streams. -static STREAMS: Lazy>>> = Lazy::new(Default::default); +static STREAMS: Lazy>> = Lazy::new(Default::default); /// The stream associated with a repository. type Stream = Box; /// Git transport protocol over an I/O stream. #[derive(Clone)] -pub struct Smart { - /// The underlying active streams, keyed by repository identifier. - streams: Arc>>, -} - -impl Smart { - /// Get access to the radicle smart transport protocol. - /// The returned object has mutable access to the underlying stream map, and is safe to clone. - pub fn singleton() -> Self { - Self { - streams: STREAMS.clone(), - } - } - - /// Take a stream from the map. - /// This makes the stream unavailable until it is re-inserted. - pub fn take(&self, id: &Id) -> Option { - #[allow(clippy::unwrap_used)] - self.streams.lock().unwrap().remove(id) - } - - pub fn insert(&self, id: Id, stream: Stream) { - #[allow(clippy::unwrap_used)] - self.streams.lock().unwrap().insert(id, stream); - } -} +struct Smart; impl git2::transport::SmartSubtransport for Smart { /// Run a git service on this transport. /// - /// Based on the URL, which must be of the form `rad://zP1GztjSdYNHK7jpdrXbaJ6Ki2Ke`, + /// Based on the URL, which must be a valid [`Url`], /// we retrieve an underlying stream and return it. /// /// We only support the upload-pack service, since only fetches are authorized by the @@ -79,17 +53,12 @@ impl git2::transport::SmartSubtransport for Smart { action: git2::transport::Service, ) -> Result, git2::Error> { let url = Url::from_str(url).map_err(|e| git2::Error::from_str(e.to_string().as_str()))?; + let mut streams = STREAMS.lock().expect("lock isn't poisoned"); - if let Some(stream) = self.take(&url.repo) { + if let Some(stream) = streams.remove(&url.node) { match action { - git2::transport::Service::UploadPackLs => {} - git2::transport::Service::UploadPack => {} - git2::transport::Service::ReceivePack => { - return Err(git2::Error::from_str( - "git-receive-pack is not supported with the custom transport", - )); - } - git2::transport::Service::ReceivePackLs => { + git2::transport::Service::UploadPackLs | git2::transport::Service::UploadPack => {} + git2::transport::Service::ReceivePack | git2::transport::Service::ReceivePackLs => { return Err(git2::Error::from_str( "git-receive-pack is not supported with the custom transport", )); @@ -98,8 +67,8 @@ impl git2::transport::SmartSubtransport for Smart { Ok(stream) } else { Err(git2::Error::from_str(&format!( - "repository {} does not have an associated stream", - url.repo + "node {} does not have an associated stream", + url.node ))) } } @@ -111,21 +80,21 @@ impl git2::transport::SmartSubtransport for Smart { /// Register the radicle transport with `git`. /// -/// Returns an error if called more than once. +/// This function can be called more than once. Only one transport will be registered. /// -pub fn register() -> Result<(), git2::Error> { - static REGISTERED: atomic::AtomicBool = atomic::AtomicBool::new(false); +pub fn register(node: RemoteId, stream: impl SmartSubtransportStream) { + static REGISTER: Once = Once::new(); // Registration is not thread-safe, so make sure we prevent re-entrancy. - if !REGISTERED.swap(true, atomic::Ordering::SeqCst) { - unsafe { - git2::transport::register(Url::SCHEME, move |remote| { - git2::transport::Transport::smart(remote, false, Smart::singleton()) - }) - } - } else { - Err(git2::Error::from_str( - "custom git transport is already registered", - )) - } + REGISTER.call_once(|| unsafe { + git2::transport::register(Url::SCHEME, move |remote| { + git2::transport::Transport::smart(remote, false, Smart) + }) + .expect("remote transport registration"); + }); + + STREAMS + .lock() + .expect("lock isn't poisoned") + .insert(node, Box::new(stream)); } diff --git a/radicle/src/storage/git/transport/remote/mock.rs b/radicle/src/storage/git/transport/remote/mock.rs new file mode 100644 index 00000000..7c5c3b17 --- /dev/null +++ b/radicle/src/storage/git/transport/remote/mock.rs @@ -0,0 +1,85 @@ +//! Mock git transport used for mocking the remote transport in tests. +use std::collections::HashMap; +use std::path::{Path, PathBuf}; +use std::str::FromStr; +use std::sync::{Mutex, Once}; +use std::{process, thread}; + +use once_cell::sync::Lazy; + +use super::Url; +use crate::storage::git::transport::ChildStream; +use crate::storage::RemoteId; + +/// Nodes registered with the mock transport. +static NODES: Lazy>> = Lazy::new(|| Mutex::new(HashMap::new())); + +/// The mock transport. +#[derive(Default)] +struct MockTransport; + +impl git2::transport::SmartSubtransport for MockTransport { + fn action( + &self, + url: &str, + service: git2::transport::Service, + ) -> Result, git2::Error> { + let url = Url::from_str(url).map_err(|e| git2::Error::from_str(e.to_string().as_str()))?; + let nodes = NODES.lock().expect("lock cannot be poisoned"); + let storage = if let Some(storage) = nodes.get(&url.node) { + match service { + git2::transport::Service::ReceivePack | git2::transport::Service::ReceivePackLs => { + return Err(git2::Error::from_str( + "git-receive-pack is not supported with the mock transport", + )); + } + _ => {} + } + storage + } else { + return Err(git2::Error::from_str(&format!( + "node {} was not registered with the mock transport", + url.node + ))); + }; + let git_dir = storage.join(url.repo.to_human()); + let mut cmd = process::Command::new("git"); + let mut child = cmd + .arg("upload-pack") + .arg("--strict") + .arg(&git_dir) + .stdin(process::Stdio::piped()) + .stdout(process::Stdio::piped()) + .stderr(process::Stdio::inherit()) + .spawn() + .expect("the `git` command is available"); + + let stdin = child.stdin.take().expect("stdin is safe to take"); + let stdout = child.stdout.take().expect("stdout is safe to take"); + + thread::spawn(move || child.wait()); + + Ok(Box::new(ChildStream { stdout, stdin })) + } + + fn close(&self) -> Result<(), git2::Error> { + Ok(()) + } +} + +/// Register a new node with the given storage path. +pub fn register(node: &RemoteId, path: &Path) { + static REGISTER: Once = Once::new(); + + REGISTER.call_once(|| unsafe { + git2::transport::register(Url::SCHEME, move |remote| { + git2::transport::Transport::smart(remote, false, MockTransport::default()) + }) + .expect("transport registration is successful"); + }); + + NODES + .lock() + .expect("the lock isn't poisoned") + .insert(*node, path.to_owned()); +} diff --git a/radicle/src/storage/git/transport/remote/url.rs b/radicle/src/storage/git/transport/remote/url.rs index 1f0e718a..2d160b79 100644 --- a/radicle/src/storage/git/transport/remote/url.rs +++ b/radicle/src/storage/git/transport/remote/url.rs @@ -21,13 +21,13 @@ pub enum UrlError { #[error("unsupported scheme: expected `heartwood://`")] UnsupportedScheme, /// Invalid node identifier. - #[error("node: {0}")] + #[error("invalid node: {0}")] InvalidNode(#[source] crypto::PublicKeyError), /// Invalid repository identifier. - #[error("repo: {0}")] + #[error("invalid repo: {0}")] InvalidRepository(#[source] IdError), /// Invalid namespace. - #[error("namespace: {0}")] + #[error("invalid namespace: {0}")] InvalidNamespace(#[source] crypto::PublicKeyError), } @@ -35,7 +35,7 @@ pub enum UrlError { /// /// `heartwood:///[/]` /// -#[derive(Debug)] +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)] pub struct Url { /// Node identifier. pub node: NodeId, diff --git a/radicle/src/test/fixtures.rs b/radicle/src/test/fixtures.rs index 5c2f1a6b..691c54e6 100644 --- a/radicle/src/test/fixtures.rs +++ b/radicle/src/test/fixtures.rs @@ -4,15 +4,18 @@ use crate::crypto::{Signer, Verified}; use crate::git; use crate::identity::Id; use crate::rad; +use crate::storage::git::transport; use crate::storage::git::Storage; use crate::storage::refs::SignedRefs; -use crate::storage::WriteStorage; /// Create a new storage with a project. pub fn storage, G: Signer>(path: P, signer: G) -> Result { let path = path.as_ref(); let storage = Storage::open(path.join("storage"))?; + transport::local::register(storage.clone()); + transport::remote::mock::register(signer.public_key(), storage.path()); + for (name, desc) in [ ("acme", "Acme's repository"), ("vim", "A text editor"), @@ -33,11 +36,13 @@ pub fn storage, G: Signer>(path: P, signer: G) -> Result, S: WriteStorage, G: Signer>( +pub fn project, G: Signer>( path: P, - storage: &S, + storage: &Storage, signer: G, ) -> Result<(Id, SignedRefs, git2::Repository, git2::Oid), rad::InitError> { + transport::local::register(storage.clone()); + let (repo, head) = repository(path); let (id, refs) = rad::init( &repo, diff --git a/radicle/src/test/storage.rs b/radicle/src/test/storage.rs index c6beceac..1796c8f9 100644 --- a/radicle/src/test/storage.rs +++ b/radicle/src/test/storage.rs @@ -2,7 +2,6 @@ use std::collections::HashMap; use std::path::{Path, PathBuf}; use git_ref_format as fmt; -use git_url::Url; use radicle_git_ext as git_ext; use crate::crypto::{Signer, Verified}; @@ -38,14 +37,6 @@ impl ReadStorage for MockStorage { self.path.as_path() } - fn url(&self, _proj: &Id) -> Url { - Url { - scheme: git_url::Scheme::Radicle, - host: Some("mock".to_string()), - ..Url::default() - } - } - fn get( &self, _remote: &RemoteId, @@ -65,10 +56,6 @@ impl WriteStorage for MockStorage { fn repository(&self, _proj: Id) -> Result { Ok(MockRepository {}) } - - fn fetch(&self, _proj_id: Id, _remote: &Url) -> Result, FetchError> { - Ok(vec![]) - } } pub struct MockRepository {} @@ -146,7 +133,11 @@ impl ReadRepository for MockRepository { } impl WriteRepository for MockRepository { - fn fetch(&mut self, _url: &Url) -> Result, FetchError> { + fn fetch( + &mut self, + _node: &RemoteId, + _namespaces: impl Into, + ) -> Result, FetchError> { Ok(vec![]) }