storage: Rewrite temporary repositories for clones — heartwood

On Windows, all attempts to clone repositories failed with

Fetch failed for rad:… from z6Mk…: Access is denied. (os error 5)

The reason is that, other than Unix-like systems, it forbids that directories that are in use are moved.

To improve the situation, take back control over what is moved and removed exactly by implementing cleanup instead of relying on impl Drop for tempfile::TempDir.

fn lock_repository is renamed to fn temporary_repository because it does not do any locking. Also, the directory name now contains “tmp” instead of “lock”.

In the future (once Rust 1.89 is a little less cutting edge and more widely available), we may opt for actual locking via std::fs::File::lock.

     eve.connect(&bob);
     eve.routes_to(&[(acme, carol), (acme, bob.id), (acme, alice.id)]);
     bob.storage.repository(acme).unwrap().remove().unwrap(); // Cause the fetch from Bob to fail.
     bob.storage.lock_repository(acme).ok(); // Prevent repo from being re-fetched.
     bob.storage.temporary_repository(acme).ok(); // Prevent repo from being re-fetched.
     test(
         "examples/rad-clone-partial-fail.md",

 use crate::transport::{ConnectionStream, Transport};
 /// The handle used for pulling or cloning changes from a remote peer.
 pub struct Handle<S> {
 pub struct Handle<R, S> {
     pub(crate) local: PublicKey,
     pub(crate) repo: Repository,
     repo: R,
     pub(crate) allowed: Allowed,
     pub(crate) transport: Transport<S>,
     /// The set of keys we will ignore when fetching from a

     pub(crate) interrupt: Arc<AtomicBool>,
 }
 impl<S> Handle<S> {
 impl<R, S> Handle<R, S> {
     pub fn is_blocked(&self, key: &PublicKey) -> bool {
         self.blocked.is_blocked(key)
     }
     #[inline]
     pub fn local(&self) -> &PublicKey {
         &self.local
     }
     pub fn interrupt_pack_writer(&mut self) {
         self.interrupt.store(true, atomic::Ordering::Relaxed);
     }
     pub fn allowed(&self) -> Allowed {
         self.allowed.clone()
     }
     pub fn into_inner(self) -> R {
         self.repo
     }
 }
 impl<R, S> Handle<R, S>
 where
     R: AsRef<Repository>,
 {
     pub fn new(
         local: PublicKey,
         repo: Repository,
         repo: R,
         follow: Allowed,
         blocked: BlockList,
         connection: S,

     where
         S: ConnectionStream,
     {
         let git_dir = repo.backend.path().to_path_buf();
         let transport = Transport::new(git_dir, BString::from(repo.id.canonical()), connection);
         let git_dir = repo.as_ref().backend.path().to_path_buf();
         let transport = Transport::new(
             git_dir,
             BString::from(repo.as_ref().id.canonical()),
             connection,
         );
         Ok(Self {
             local,

         })
     }
     pub fn is_blocked(&self, key: &PublicKey) -> bool {
         self.blocked.is_blocked(key)
     }
     #[inline]
     pub fn repository(&self) -> &Repository {
         &self.repo
     }
     pub fn repository_mut(&mut self) -> &mut Repository {
         &mut self.repo
     }
     pub fn local(&self) -> &PublicKey {
         &self.local
     }
     pub fn interrupt_pack_writer(&mut self) {
         self.interrupt.store(true, atomic::Ordering::Relaxed);
         self.repo.as_ref()
     }
     pub fn verified(&self, head: Oid) -> Result<Doc, DocError> {
         Ok(self.repo.identity_doc_at(head)?.doc)
     }
     pub fn allowed(&self) -> Allowed {
         self.allowed.clone()
         Ok(self.repository().identity_doc_at(head)?.doc)
     }
 }

 pub use gix_protocol::{transport::bstr::ByteSlice, RemoteProgress};
 pub use handle::Handle;
 pub use policy::{Allowed, BlockList, Scope};
 use radicle::storage::git::Repository;
 pub use state::{FetchLimit, FetchResult};
 pub use transport::Transport;

 /// It is expected that the local peer has a copy of the repository
 /// and is pulling new changes. If the repository does not exist, then
 /// [`clone`] should be used.
 pub fn pull<S>(
     handle: &mut Handle<S>,
 pub fn pull<R, S>(
     handle: &mut Handle<R, S>,
     limit: FetchLimit,
     remote: PublicKey,
     refs_at: Option<Vec<RefsAt>>,
 ) -> Result<FetchResult, Error>
 where
     R: AsRef<Repository>,
     S: transport::ConnectionStream,
 {
     let start = Instant::now();

     log::debug!(
         target: "fetch",
         "Finished pull of {} ({}ms)",
         handle.repo.id(),
         handle.repository().id(),
         start.elapsed().as_millis()
     );
     result

         .run(handle, &handshake, limit, remote, None)
         .map_err(Error::Protocol);
     let elapsed = start.elapsed().as_millis();
     let rid = handle.repo.id();
     let rid = handle.repository().id();
     match &result {
         Ok(_) => {

     result
 }
 fn perform_handshake<S>(handle: &mut Handle<S>) -> Result<handshake::Outcome, Error>
 fn perform_handshake<R, S>(handle: &mut Handle<R, S>) -> Result<handshake::Outcome, Error>
 where
     S: transport::ConnectionStream,
 {

 use std::collections::{BTreeMap, BTreeSet};
 use std::ops::{Deref, Not as _};
 use radicle::storage::git::Repository;
 pub use radicle::storage::refs::SignedRefsAt;
 pub use radicle::storage::{git::Validation, Validations};
 use radicle::{crypto::PublicKey, storage::ValidateRepository};

     /// Construct a `DelegateStatus` with [`SignedRefsAt`] signed reference
     /// data, if it can be found in `repo`.
     pub fn load<S>(
     pub fn load<R, S>(
         self,
         cached: &Cached<S>,
     ) -> Result<DelegateStatus<Option<SignedRefsAt>>, radicle::storage::refs::Error> {
         cached: &Cached<R, S>,
     ) -> Result<DelegateStatus<Option<SignedRefsAt>>, radicle::storage::refs::Error>
     where
         R: AsRef<Repository>,
     {
         let remote = *self.remote();
         self.traverse(|_| cached.load(&remote))
     }

     ///
     /// If the sigrefs are missing for a given remote, regardless of delegate
     /// status, then that remote is filtered out.
     pub(crate) fn load<'a, S>(
         cached: &Cached<S>,
     pub(crate) fn load<'a, R, S>(
         cached: &Cached<R, S>,
         remotes: impl Iterator<Item = &'a PublicKey>,
     ) -> Result<Self, error::RemoteRefs> {
     ) -> Result<Self, error::RemoteRefs>
     where
         R: AsRef<Repository>,
     {
         remotes
             .filter_map(|id| match cached.load(id) {
                 Ok(None) => None,

 use radicle::prelude::Verified;
 use radicle::storage;
 use radicle::storage::git::Repository;
 use radicle::storage::refs::RefsAt;
 use radicle::storage::{
     git::Validation, Remote, RemoteId, RemoteRepository, Remotes, ValidateRepository, Validations,

         ap
     }
     pub(crate) fn as_cached<'a, S>(&'a mut self, handle: &'a mut Handle<S>) -> Cached<'a, S> {
     pub(crate) fn as_cached<'a, R, S>(
         &'a mut self,
         handle: &'a mut Handle<R, S>,
     ) -> Cached<'a, R, S> {
         Cached {
             handle,
             state: self,

 impl FetchState {
     /// Perform the ls-refs and fetch for the given `step`. The result
     /// of these processes is kept track of in the internal state.
     pub(super) fn run_stage<S, F>(
     pub(super) fn run_stage<R, S, F>(
         &mut self,
         handle: &mut Handle<S>,
         handle: &mut Handle<R, S>,
         handshake: &handshake::Outcome,
         step: &F,
     ) -> Result<BTreeSet<PublicKey>, error::Step>
     where
         R: AsRef<Repository>,
         S: transport::ConnectionStream,
         F: ProtocolStage,
     {

         log::trace!(target: "fetch", "Received refs {refs:?}");
         step.pre_validate(&refs)?;
         let wants_haves = step.wants_haves(&handle.repo, &refs)?;
         let wants_haves = step.wants_haves(handle.repository(), &refs)?;
         if !wants_haves.wants.is_empty() {
             let keepfile =
                 handle

             }
         }
         let up = step.prepare_updates(self, &handle.repo, &refs)?;
         let up = step.prepare_updates(self, handle.repository(), &refs)?;
         self.update_all(up.tips);
         Ok(fetched)

     /// The resulting [`sigrefs::RemoteRefs`] will be the set of
     /// `rad/sigrefs` of the fetched remotes.
     #[allow(clippy::too_many_arguments)]
     fn run_special_refs<S>(
     fn run_special_refs<R, S>(
         &mut self,
         handle: &mut Handle<S>,
         handle: &mut Handle<R, S>,
         handshake: &handshake::Outcome,
         delegates: BTreeSet<PublicKey>,
         threshold: usize,

                     remote,
                     data: Some(sigrefs),
                 } => {
                     if let Some(SignedRefsAt { at, .. }) = SignedRefsAt::load(remote, &handle.repo)?
                     if let Some(SignedRefsAt { at, .. }) =
                         SignedRefsAt::load(remote, handle.repository())?
                     {
                         // Prune non-delegates if they're behind or
                         // diverged. A diverged case is non-fatal for
                         // delegates.
                         if matches!(
                             repository::ancestry(&handle.repo, at, sigrefs.at)?,
                             repository::ancestry(handle.repository(), at, sigrefs.at)?,
                             repository::Ancestry::Behind | repository::Ancestry::Diverged
                         ) {
                             self.prune(&remote);

         // delegates that pass the threshold.
         if valid_delegates.len() >= threshold {
             let applied = repository::update(
                 &handle.repo,
                 handle.repository(),
                 self.tips
                     .clone()
                     .into_values()

 /// A cached version of [`Handle`] by using the underlying
 /// [`FetchState`]'s data for performing lookups.
 pub(crate) struct Cached<'a, S> {
     handle: &'a mut Handle<S>,
 pub(crate) struct Cached<'a, R, S> {
     handle: &'a mut Handle<R, S>,
     state: &'a mut FetchState,
 }
 impl<S> Cached<'_, S> {
 impl<R, S> Cached<'_, R, S>
 where
     R: AsRef<Repository>,
 {
     /// Resolves `refname` to its [`ObjectId`] by first looking at the
     /// [`FetchState`] and falling back to the [`Handle::refdb`].
     pub fn refname_to_id<'b, N>(

     {
         let refname = refname.into();
         match self.state.refs.refname_to_id(refname.clone()) {
             None => repository::refname_to_id(&self.handle.repo, refname),
             None => repository::refname_to_id(self.handle.repository(), refname),
             Some(oid) => Ok(Some(oid)),
         }
     }

     pub fn load(&self, remote: &PublicKey) -> Result<Option<SignedRefsAt>, sigrefs::error::Load> {
         match self.state.sigrefs.get(remote) {
             None => SignedRefsAt::load(*remote, &self.handle.repo),
             Some(tip) => SignedRefsAt::load_at(*tip, *remote, &self.handle.repo).map(Some),
             None => SignedRefsAt::load(*remote, self.handle.repository()),
             Some(tip) => SignedRefsAt::load_at(*tip, *remote, self.handle.repository()).map(Some),
         }
     }

     }
 }
 impl<S> RemoteRepository for Cached<'_, S> {
 impl<R, S> RemoteRepository for Cached<'_, R, S>
 where
     R: AsRef<Repository>,
 {
     fn remote(&self, remote: &RemoteId) -> Result<Remote, storage::refs::Error> {
         // N.b. this is unused so we just delegate to the underlying
         // repository for a correct implementation.
         self.handle.repo.remote(remote)
         self.handle.repository().remote(remote)
     }
     fn remotes(&self) -> Result<Remotes<Verified>, storage::refs::Error> {

     }
     fn remote_refs_at(&self) -> Result<Vec<RefsAt>, storage::refs::Error> {
         self.handle.repo.remote_refs_at()
         self.handle.repository().remote_refs_at()
     }
 }
 impl<S> ValidateRepository for Cached<'_, S> {
 impl<R, S> ValidateRepository for Cached<'_, R, S>
 where
     R: AsRef<Repository>,
 {
     // N.b. we don't verify the `rad/id` of each remote since they may
     // not have a reference to the COB if they have not interacted
     // with it.

 use radicle::identity::doc::CanonicalRefsError;
 use radicle::identity::CanonicalRefs;
 use radicle::storage::git::TempRepository;
 pub(crate) use radicle_protocol::worker::fetch::error;
 use std::collections::BTreeSet;

 pub enum Handle {
     Clone {
         handle: radicle_fetch::Handle<ChannelsFlush>,
         tmp: tempfile::TempDir,
         handle: radicle_fetch::Handle<TempRepository, ChannelsFlush>,
     },
     Pull {
         handle: radicle_fetch::Handle<ChannelsFlush>,
         handle: radicle_fetch::Handle<Repository, ChannelsFlush>,
         notifications: node::notifications::StoreWriter,
     },
 }

                 notifications,
             })
         } else {
             let (repo, tmp) = storage.lock_repository(rid)?;
             let repo = storage.temporary_repository(rid)?;
             let handle = radicle_fetch::Handle::new(local, repo, follow, blocked, channels)?;
             Ok(Handle::Clone { handle, tmp })
             Ok(Handle::Clone { handle })
         }
     }

         refs_at: Option<Vec<RefsAt>>,
     ) -> Result<FetchResult, error::Fetch> {
         let (result, clone, notifs) = match self {
             Self::Clone { mut handle, tmp } => {
             Self::Clone { mut handle } => {
                 log::debug!(target: "worker", "{} cloning from {remote}", handle.local());
                 let result = radicle_fetch::clone(&mut handle, limit, remote)?;
                 mv(tmp, storage, &rid)?;
                 (result, true, None)
                 match radicle_fetch::clone(&mut handle, limit, remote) {
                     Err(err) => {
                         handle.into_inner().cleanup();
                         return Err(err.into());
                     }
                     Ok(result) => {
                         handle.into_inner().mv(storage.path_of(&rid))?;
                         (result, true, None)
                     }
                 }
             }
             Self::Pull {
                 mut handle,

     }
 }
 /// In the case of cloning, we have performed the fetch into a
 /// temporary directory -- ensuring that no concurrent operations
 /// see an empty repository.
 ///
 /// At the end of the clone, we perform a rename of the temporary
 /// directory to the storage repository.
 ///
 /// # Errors
 ///   - Will fail if `storage` contains `rid` already.
 fn mv(tmp: tempfile::TempDir, storage: &Storage, rid: &RepoId) -> Result<(), error::Fetch> {
     use std::io::{Error, ErrorKind};
     let from = tmp.path();
     let to = storage.path_of(rid);
     if !to.exists() {
         std::fs::rename(from, to)?;
     } else {
         log::warn!(target: "worker", "Refusing to move cloned repository {rid} already exists");
         return Err(Error::new(
             ErrorKind::AlreadyExists,
             format!("repository already exists {to:?}"),
         )
         .into());
     }
     Ok(())
 }
 // Post notifications for the given refs.
 fn notify(
     rid: &RepoId,

 [features]
 default = []
 test = ["qcheck", "radicle-crypto/test"]
 test = ["tempfile", "qcheck", "radicle-crypto/test"]
 logger = ["colored", "chrono"]
 [dependencies]

 serde-untagged = "0.1.7"
 siphasher = "1.0.0"
 sqlite = { workspace = true, features = ["bundled"] }
 tempfile = { workspace = true }
 tempfile = { workspace = true, optional = true }
 thiserror = { workspace = true }
 unicode-normalization = { version = "0.1" }

 pub mod cob;
 pub mod transport;
 pub mod temp;
 pub use temp::TempRepository;
 use std::collections::{BTreeMap, BTreeSet, HashMap};
 use std::ops::{Deref, DerefMut};
 use std::path::{Path, PathBuf};

 use std::{fs, io};
 use crypto::Verified;
 use tempfile::TempDir;
 use crate::git::canonical::Quorum;
 use crate::identity::crefs::GetCanonicalRefs as _;