diff --git a/radicle-node/src/runtime.rs b/radicle-node/src/runtime.rs index 40533e1c..2c424b97 100644 --- a/radicle-node/src/runtime.rs +++ b/radicle-node/src/runtime.rs @@ -156,6 +156,7 @@ impl Runtime { } let pool = worker::Pool::with( + id, worker_recv, handle.clone(), worker::Config { diff --git a/radicle-node/src/service.rs b/radicle-node/src/service.rs index b5ec188f..5c75b9b6 100644 --- a/radicle-node/src/service.rs +++ b/radicle-node/src/service.rs @@ -1072,6 +1072,7 @@ where return Ok(()); } } + // Accept the request and instruct the transport to handover the socket to the worker. self.reactor.write(peer, Message::FetchOk { rid }); self.reactor.fetch(peer, rid, FetchDirection::Responder); @@ -1092,14 +1093,16 @@ where } debug!(target: "service", "Fetch accepted for {rid} from {remote}.."); + let namespaces = if let Ok(_repo) = self.storage.repository(rid) { + // FIXME(finto): using remotes breaks test_gossip_during_fetch, so we use default for now + Namespaces::default() + } else { + Namespaces::default() + }; + // Instruct the transport to handover the socket to the worker. - self.reactor.fetch( - peer, - rid, - FetchDirection::Initiator { - namespaces: Namespaces::default(), - }, - ); + self.reactor + .fetch(peer, rid, FetchDirection::Initiator { namespaces }); } (session::State::Connecting { .. }, msg) => { error!(target: "service", "Received {:?} from connecting peer {}", msg, peer.id); @@ -1198,6 +1201,21 @@ where // SAFETY: `REF_REMOTE_LIMIT` is greater than 1, thus the bounded vec can hold at least // one remote. .unwrap(), + Namespaces::Many(pks) => { + for remote_id in pks.into_iter() { + if refs + .push((*remote_id, repo.remote(&remote_id)?.refs.unverified())) + .is_err() + { + warn!( + target: "service", + "refs announcement limit ({}) exceeded, peers will see only some of your repository references", + REF_REMOTE_LIMIT, + ); + break; + } + } + } } let msg = AnnouncementMessage::from(RefsAnnouncement { diff --git a/radicle-node/src/service/session.rs b/radicle-node/src/service/session.rs index d1131509..1487becf 100644 --- a/radicle-node/src/service/session.rs +++ b/radicle-node/src/service/session.rs @@ -108,6 +108,8 @@ pub enum Error { Timeout, #[error("handshake error")] Handshake(String), + #[error("failed to inspect remotes for fetch: {0}")] + Remotes(#[from] storage::refs::Error), } /// A peer session. Each connected peer will have one session. diff --git a/radicle-node/src/test/simulator.rs b/radicle-node/src/test/simulator.rs index d47dd68c..2e3b1c1c 100644 --- a/radicle-node/src/test/simulator.rs +++ b/radicle-node/src/test/simulator.rs @@ -679,6 +679,7 @@ fn fetch( let namespace = match namespaces.into() { Namespaces::All => None, Namespaces::One(ns) => Some(ns), + Namespaces::Many(ns) => Some(ns.head), }; let mut updates = Vec::new(); let mut callbacks = git::RemoteCallbacks::new(); diff --git a/radicle-node/src/worker.rs b/radicle-node/src/worker.rs index 0032f255..0fe5a1fe 100644 --- a/radicle-node/src/worker.rs +++ b/radicle-node/src/worker.rs @@ -7,7 +7,7 @@ use cyphernet::Ecdh; use netservices::tunnel::Tunnel; use netservices::{NetSession, SplitIo}; -use radicle::crypto::Signer; +use radicle::crypto::{PublicKey, Signer}; use radicle::identity::{Id, IdentityError}; use radicle::storage::{Namespaces, ReadRepository, RefUpdate, WriteRepository, WriteStorage}; use radicle::{git, Storage}; @@ -37,6 +37,8 @@ pub struct Config { /// Error returned by fetch. #[derive(thiserror::Error, Debug)] pub enum FetchError { + #[error("the 'git fetch' command failed with exit code '{code}'")] + CommandFailed { code: i32 }, #[error(transparent)] Git(#[from] git::raw::Error), #[error(transparent)] @@ -73,6 +75,7 @@ pub struct TaskResult { /// A worker that replicates git objects. struct Worker { + local: PublicKey, storage: Storage, tasks: chan::Receiver>, daemon: net::SocketAddr, @@ -197,15 +200,22 @@ impl Worker { // a state we can't roll back. cmd.arg("--prune"); } + Namespaces::Many(_) => { + // Same case as All + } } if self.atomic { // Enable atomic fetch. Only works with Git 2.31 and later. cmd.arg("--atomic"); } + + // Ignore our own remote when fetching + let mut fetchspecs = namespaces.as_fetchspecs(); + fetchspecs.push(format!("^refs/namespaces/{}/*", self.local)); + cmd.arg(format!("git://{tunnel_addr}/{}", repo.id.canonical())) - // FIXME: We need to omit our own namespace from this refspec in case we're fetching '*'. - .arg(namespaces.as_fetchspec()) + .args(&fetchspecs) .stdout(process::Stdio::piped()) .stderr(process::Stdio::piped()) .stdin(process::Stdio::piped()); @@ -224,16 +234,18 @@ impl Worker { let _ = tunnel.tunnel_once(popol::Poller::new(), self.timeout)?; // TODO: Parse fetch output to return updates. - if child.wait()?.success() { + let result = child.wait()?; + if result.success() { log::debug!(target: "worker", "Fetch for {} exited successfully", rid); + let head = repo.set_head()?; + log::debug!(target: "worker", "Head for {} set to {head}", rid); + Ok(vec![]) } else { log::error!(target: "worker", "Fetch for {} failed", rid); + Err(FetchError::CommandFailed { + code: result.code().unwrap_or(1), + }) } - let head = repo.set_head()?; - - log::debug!(target: "worker", "Head for {} set to {head}", rid); - - Ok(vec![]) } fn upload_pack( @@ -314,6 +326,7 @@ pub struct Pool { impl Pool { /// Create a new worker pool with the given parameters. pub fn with( + local: PublicKey, tasks: chan::Receiver>, handle: Handle, config: Config, @@ -321,6 +334,7 @@ impl Pool { let mut pool = Vec::with_capacity(config.capacity); for _ in 0..config.capacity { let worker = Worker { + local, tasks: tasks.clone(), handle: handle.clone(), storage: config.storage.clone(), diff --git a/radicle/src/storage.rs b/radicle/src/storage.rs index c1429974..f3cf3944 100644 --- a/radicle/src/storage.rs +++ b/radicle/src/storage.rs @@ -36,13 +36,35 @@ pub enum Namespaces { All, /// A single namespace, by public key. One(PublicKey), + /// Many namespaces, by public keys. + Many(NonEmpty), } impl Namespaces { - pub fn as_fetchspec(&self) -> String { + pub fn remotes(repo: &R) -> Result, refs::Error> + where + R: ReadRepository, + { + Ok(NonEmpty::collect(repo.remotes()?.keys().copied()).map(Self::Many)) + } + + pub fn delegates(repo: &R) -> Result + where + R: ReadRepository, + { + Ok(Self::Many(repo.delegates()?.map(PublicKey::from))) + } + + pub fn as_fetchspecs(&self) -> Vec { match self { - Self::All => String::from("refs/namespaces/*:refs/namespaces/*"), - Self::One(pk) => format!("refs/namespaces/{pk}/refs/*:refs/namespaces/{pk}/refs/*"), + Self::All => vec![String::from("refs/namespaces/*:refs/namespaces/*")], + Self::One(pk) => vec![format!( + "refs/namespaces/{pk}/refs/*:refs/namespaces/{pk}/refs/*" + )], + Self::Many(pks) => pks + .iter() + .map(|pk| format!("refs/namespaces/{pk}/refs/*:refs/namespaces/{pk}/refs/*")) + .collect(), } } } @@ -53,6 +75,12 @@ impl From for Namespaces { } } +impl From> for Namespaces { + fn from(pks: NonEmpty) -> Self { + Self::Many(pks) + } +} + /// Storage error. #[derive(Error, Debug)] pub enum Error {