From e937cdd79acca6178e54e365c491f2d9de2b0268 Mon Sep 17 00:00:00 2001 From: Fintan Halpenny Date: Fri, 24 Feb 2023 11:18:15 +0000 Subject: [PATCH] node: improve fetch The fetch implementation will potentially fetch the local operator's remote, which is unnecessary and could result in deleted data. Fix this by threading the local operator's public key and add an ignore refspec to the fetch arguments, i.e. ^refs/namespaces//* The implementation also strictly fetches all or one namespace. This is improved upon by introducing a new variant for specifying a set of namespaces. Signed-off-by: Fintan Halpenny X-Clacks-Overhead: GNU Terry Pratchett --- radicle-node/src/runtime.rs | 1 + radicle-node/src/service.rs | 32 +++++++++++++++++++++------ radicle-node/src/service/session.rs | 2 ++ radicle-node/src/test/simulator.rs | 1 + radicle-node/src/worker.rs | 32 +++++++++++++++++++-------- radicle/src/storage.rs | 34 ++++++++++++++++++++++++++--- 6 files changed, 83 insertions(+), 19 deletions(-) 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 {