radicle-fetch: More fine-grained errors — heartwood

The fn io_error disguises all sorts of errors as I/O errors, making errors in the transport component of radicle-fetch harder to understand.

Instead, expose a new error type that allows to discriminate.

 Fetching rad:z2ug5mwNKZB8KGpBDRTrWHAMbvHCu from the network, found 1 potential seed(s).
 ✗ Target not met: could not fetch from [z6MknSLrJoTcukLrE435hVNQT4JUhbvWLX4kUzqkEStBU8Vi], and required 1 more seed(s)
 ! Warning: Failed to fetch from 1 seed(s).
 ! Warning: z6MknSLrJoTcukLrE435hVNQT4JUhbvWLX4kUzqkEStBU8Vi: failed to perform fetch handshake
 ! Warning: z6MknSLrJoTcukLrE435hVNQT4JUhbvWLX4kUzqkEStBU8Vi: failed to perform fetch handshake: [..]
 ✗ Error: no seeds found for rad:z2ug5mwNKZB8KGpBDRTrWHAMbvHCu
 ```

 }
 pub mod error {
     use std::io;
     use radicle::node::policy;
     use radicle::prelude::RepoId;
     use radicle::{git, storage};

     #[derive(Debug, Error)]
     pub enum Init {
         #[error(transparent)]
         Io(#[from] io::Error),
         #[error(transparent)]
         Tracking(#[from] policy::config::Error),
     }

 mod stage;
 mod state;
 use std::io;
 use std::time::Instant;
 use gix_protocol::handshake;

 #[derive(Debug, Error)]
 pub enum Error {
     #[error("failed to perform fetch handshake")]
     Handshake {
         #[source]
         err: io::Error,
     },
     #[error("failed to perform fetch handshake: {0}")]
     Handshake(#[from] Box<handshake::Error>),
     #[error("failed to load `rad/id`")]
     Identity {
         #[source]

 where
     S: transport::ConnectionStream,
 {
     handle.transport.handshake().map_err(|err| {
     let result = handle.transport.handshake();
     if let Err(err) = &result {
         log::warn!(target: "fetch", "Failed to perform handshake: {err}");
         Error::Handshake { err }
     })
     }
     Ok(result?)
 }

 pub const DEFAULT_FETCH_DATA_REFS_LIMIT: u64 = 1024 * 1024 * 1024 * 5;
 pub mod error {
     use std::io;
     use radicle::git::Oid;
     use radicle::prelude::PublicKey;
     use thiserror::Error;
     use crate::{git, git::repository, handle, sigrefs, stage};
     use crate::{git, git::repository, handle, sigrefs, stage, transport};
     #[derive(Debug, Error)]
     pub enum Step {
         #[error(transparent)]
         Io(#[from] io::Error),
         #[error(transparent)]
         Layout(#[from] stage::error::Layout),
         #[error(transparent)]
         Prepare(#[from] stage::error::Prepare),
         #[error(transparent)]
         WantsHaves(#[from] stage::error::WantsHaves),
         #[error(transparent)]
         Transport(#[from] transport::Error),
     }
     #[derive(Debug, Error)]

             current: Oid,
             received: Oid,
         },
         #[error(transparent)]
         Io(#[from] io::Error),
         #[error("canonical 'refs/rad/id' is missing")]
         MissingRadId,
         #[error(transparent)]

 pub trait ConnectionStream {
     type Read: io::Read;
     type Write: io::Write + SignalEof;
     type Error: std::error::Error + Send + Sync + 'static;
     fn open(&mut self) -> Result<(&mut Self::Read, &mut Self::Write), Self::Error>;
     fn open(&mut self) -> (&mut Self::Read, &mut Self::Write);
 }
 /// The ability to signal EOF to the server side so that it can stop
 /// serving for this fetch request.
 pub trait SignalEof {
     type Error: std::error::Error + Send + Sync + 'static;
     /// Since the git protocol is tunneled over an existing
     /// connection, we can't signal the end of the protocol via the
     /// usual means, which is to close the connection. Git also

     /// the git protocol. This message can then be processed by the
     /// remote worker to end the protocol. We use the special "eof"
     /// control message for this.
     fn eof(&mut self) -> Result<(), Self::Error>;
     fn eof(&mut self) -> io::Result<()>;
 }
 /// Configuration for running a Git `handshake`, `ls-refs`, or

     stream: S,
 }
 #[derive(Error, Debug)]
 pub enum Error {
     #[error("gix ls-refs error: {0}")]
     LsRefs(#[from] gix_protocol::ls_refs::Error),
     #[error("gix fetch error: {0}")]
     Fetch(#[from] gix_protocol::fetch::Error),
     #[error("empty or no packfile received")]
     Empty,
     #[error("wanted object not found: {0}")]
     NotFound(Oid),
     #[error("gix pack index error: {0}")]
     PackIndex(#[from] gix_pack::index::init::Error),
 }
 impl<S> Transport<S>
 where
     S: ConnectionStream,

     }
     /// Perform the handshake with the server side.
     pub(crate) fn handshake(&mut self) -> io::Result<handshake::Outcome> {
     pub(crate) fn handshake(&mut self) -> Result<handshake::Outcome, Box<handshake::Error>> {
         log::trace!(target: "fetch", "Performing handshake for {}", self.repo);
         let (read, write) = self.stream.open().map_err(io_other)?;
         let (read, write) = self.stream.open();
         gix_protocol::fetch::handshake(
             &mut Connection::new(read, write, self.repo.clone()),
             |_| Ok(None),
             vec![],
             &mut progress::Discard,
         )
         .map_err(io_other)
         .map_err(Box::new)
     }
     /// Perform ls-refs with the server side.

         &mut self,
         mut prefixes: Vec<BString>,
         handshake: &handshake::Outcome,
     ) -> io::Result<Vec<handshake::Ref>> {
     ) -> Result<Vec<handshake::Ref>, Error> {
         prefixes.sort();
         prefixes.dedup();
         let (read, write) = self.stream.open().map_err(io_other)?;
         ls_refs::run(
         let (read, write) = self.stream.open();
         Ok(ls_refs::run(
             ls_refs::Config {
                 prefixes,
                 repo: self.repo.clone(),

         wants_haves: WantsHaves,
         interrupt: Arc<AtomicBool>,
         handshake: &handshake::Outcome,
     ) -> io::Result<Option<Keepfile>> {
     ) -> Result<Option<Keepfile>, Error> {
         log::trace!(
             target: "fetch",
             "Running fetch wants={:?}, haves={:?}",

             wants_haves.haves
         );
         let out = {
             let (read, write) = self.stream.open().map_err(io_other)?;
             let (read, write) = self.stream.open();
             fetch::run(
                 wants_haves.clone(),
                 fetch::PackWriter {

                 handshake,
                 Connection::new(read, write, self.repo.clone()),
                 &mut progress::Discard,
             )
             .map_err(io_other)?
             )?
         };
         let pack_path = out
             .pack
             .ok_or_else(|| {
                 io::Error::new(
                     io::ErrorKind::UnexpectedEof,
                     "empty or no packfile received",
                 )
             })?
             .ok_or(Error::Empty)?
             .index_path
             .expect("written packfile must have a path");

         {
             use gix_pack::index::File;
             let idx = File::at(pack_path, gix_hash::Kind::Sha1).map_err(io_other)?;
             let idx = File::at(pack_path, gix_hash::Kind::Sha1)?;
             for oid in wants_haves.wants {
                 if idx.lookup(oid::to_object_id(oid)).is_none() {
                     return Err(io::Error::new(
                         io::ErrorKind::NotFound,
                         format!("wanted {oid} not found in pack"),
                     ));
                     return Err(Error::NotFound(oid));
                 }
             }
         }

     /// Signal to the server side that we are done sending ls-refs and
     /// fetch commands.
     pub(crate) fn done(&mut self) -> io::Result<()> {
         let (_, w) = self.stream.open().map_err(io_other)?;
         w.eof().map_err(io_other)
         let (_, w) = self.stream.open();
         w.eof()
     }
 }

     }
 }
 fn io_other(err: impl std::error::Error + Send + Sync + 'static) -> io::Error {
     io::Error::other(err)
 }
 #[derive(Debug, Error)]
 pub enum WantsHavesError {
     #[error(transparent)]

     }
 }
 fn agent_name() -> io::Result<String> {
     Ok(format!(
         "git/{}",
         radicle::git::version().map_err(io_other)?
     ))
 fn agent_name() -> String {
     let version = match radicle::git::version() {
         Ok(version) => version,
         Err(err) => {
             use radicle::git::VERSION_REQUIRED;
             log::warn!(target: "fetch", "The git version could not be determined: {err}");
             log::warn!(target: "fetch", "Pretending that we are on git version {VERSION_REQUIRED}.");
             VERSION_REQUIRED
         }
     };
     format!("git/{version}")
 }

             use_multi_pack_index: true,
             current_dir: Some(self.git_dir.clone()),
         };
         let thickener = Arc::new(gix_odb::Store::at_opts(
             self.git_dir.join("objects"),
             &mut [].into_iter(),
             odb_opts,
         )?);
         let thickener = Arc::new(
             gix_odb::Store::at_opts(self.git_dir.join("objects"), &mut [].into_iter(), odb_opts)
                 .map_err(error::PackWriter::StoreOpen)?,
         );
         let thickener = thickener.to_handle_arc();
         Ok(pack::Bundle::write_to_directory(
             pack,

         keepfile: None,
     };
     let mut negotiate = Negotiate { wants_haves };
     let agent = agent_name().map_err(Error::ReadRemainingBytes)?;
     let agent = agent_name();
     let mut pack_out = None;
     let mut handshake = handshake.clone();

                 arg.push_str(prefix);
                 args.push(arg)
             }
             features.push(("agent", Some(Cow::Owned(agent_name()?))));
             features.push(("agent", Some(Cow::Owned(agent_name()))));
             Ok(gix_protocol::ls_refs::Action::Continue)
         },
         progress,

 use std::convert::Infallible;
 use std::io::{Read, Write};
 use std::ops::Deref;
 use std::{fmt, io, time};

 impl radicle_fetch::transport::ConnectionStream for ChannelsFlush {
     type Read = ChannelReader;
     type Write = ChannelFlushWriter;
     type Error = Infallible;
     fn open(&mut self) -> Result<(&mut Self::Read, &mut Self::Write), Self::Error> {
         Ok((&mut self.receiver, &mut self.sender))
     fn open(&mut self) -> (&mut Self::Read, &mut Self::Write) {
         (&mut self.receiver, &mut self.sender)
     }
 }

 }
 impl radicle_fetch::transport::SignalEof for ChannelFlushWriter<Vec<u8>> {
     type Error = io::Error;
     fn eof(&mut self) -> io::Result<()> {
         self.writer.send(ChannelEvent::Eof)?;
         self.flush()