radicle-fetch: More fine-grained errors

The `fn io_other` 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.

crates/radicle-cli/examples/rad-init-private-clone.md

@@ -10,7 +10,7 @@ $ rad clone rad:z2ug5mwNKZB8KGpBDRTrWHAMbvHCu --seed z6MknSLrJoTcukLrE435hVNQT4J
 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
 ```
 

crates/radicle-fetch/src/handle.rs

@@ -83,8 +83,6 @@ impl<S> Handle<S> {
 }
 
 pub mod error {
-    use std::io;
-
     use radicle::node::policy;
     use radicle::prelude::RepoId;
     use radicle::{git, storage};
@@ -92,8 +90,6 @@ pub mod error {
 
     #[derive(Debug, Error)]
     pub enum Init {
-        #[error(transparent)]
-        Io(#[from] io::Error),
         #[error(transparent)]
         Tracking(#[from] policy::config::Error),
     }

crates/radicle-fetch/src/lib.rs

@@ -9,7 +9,6 @@ mod refs;
 mod stage;
 mod state;
 
-use std::io;
 use std::time::Instant;
 
 use gix_protocol::handshake;
@@ -28,11 +27,8 @@ use thiserror::Error;
 
 #[derive(Debug, Error)]
 pub enum Error {
-    #[error("failed to perform fetch handshake")]
+    #[error("failed to perform fetch handshake: {0}")]
-    Handshake {
+    Handshake(#[from] Box<handshake::Error>),
-        #[source]
-        err: io::Error,
-    },
     #[error("failed to load `rad/id`")]
     Identity {
         #[source]
@@ -128,8 +124,11 @@ fn perform_handshake<S>(handle: &mut Handle<S>) -> Result<handshake::Outcome, Er
 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?)
 }

crates/radicle-fetch/src/state.rs

@@ -30,24 +30,22 @@ pub const DEFAULT_FETCH_SPECIAL_REFS_LIMIT: u64 = 1024 * 1024 * 5;
 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)]
@@ -62,8 +60,6 @@ pub mod error {
             current: Oid,
             received: Oid,
         },
-        #[error(transparent)]
-        Io(#[from] io::Error),
         #[error("canonical 'refs/rad/id' is missing")]
         MissingRadId,
         #[error(transparent)]

crates/radicle-fetch/src/transport.rs

@@ -27,16 +27,13 @@ use crate::git::repository;
 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
@@ -48,7 +45,7 @@ pub trait SignalEof {
     /// 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
@@ -59,6 +56,20 @@ pub struct Transport<S> {
     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,
@@ -79,16 +90,16 @@ where
     }
 
     /// 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.
@@ -96,11 +107,11 @@ where
         &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)?;
+        let (read, write) = self.stream.open();
-        ls_refs::run(
+        Ok(ls_refs::run(
             ls_refs::Config {
                 prefixes,
                 repo: self.repo.clone(),
@@ -108,8 +119,7 @@ where
             handshake,
             Connection::new(read, write, self.repo.clone()),
             &mut progress::Discard,
-        )
+        )?)
-        .map_err(io_other)
     }
 
     /// Perform the fetch with the server side.
@@ -118,7 +128,7 @@ where
         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={:?}",
@@ -126,7 +136,7 @@ where
             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 {
@@ -136,17 +146,11 @@ where
                 handshake,
                 Connection::new(read, write, self.repo.clone()),
                 &mut progress::Discard,
-            )
+            )?
-            .map_err(io_other)?
         };
         let pack_path = out
             .pack
-            .ok_or_else(|| {
+            .ok_or(Error::Empty)?
-                io::Error::new(
-                    io::ErrorKind::UnexpectedEof,
-                    "empty or no packfile received",
-                )
-            })?
             .index_path
             .expect("written packfile must have a path");
 
@@ -157,13 +161,10 @@ where
         {
             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(
+                    return Err(Error::NotFound(oid));
-                        io::ErrorKind::NotFound,
-                        format!("wanted {oid} not found in pack"),
-                    ));
                 }
             }
         }
@@ -174,8 +175,8 @@ where
     /// 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)?;
+        let (_, w) = self.stream.open();
-        w.eof().map_err(io_other)
+        w.eof()
     }
 }
 
@@ -251,10 +252,6 @@ where
     }
 }
 
-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)]

crates/radicle-fetch/src/transport/fetch.rs

@@ -22,8 +22,8 @@ pub mod error {
 
     #[derive(Debug, Error)]
     pub enum PackWriter {
-        #[error(transparent)]
+        #[error("i/o error opening store: {0}")]
-        Io(#[from] io::Error),
+        StoreOpen(#[from] io::Error),
         #[error(transparent)]
         Write(#[from] gix_pack::bundle::write::Error),
     }
@@ -61,11 +61,10 @@ impl PackWriter {
             use_multi_pack_index: true,
             current_dir: Some(self.git_dir.clone()),
         };
-        let thickener = Arc::new(gix_odb::Store::at_opts(
+        let thickener = Arc::new(
-            self.git_dir.join("objects"),
+            gix_odb::Store::at_opts(self.git_dir.join("objects"), &mut [].into_iter(), odb_opts)
-            &mut [].into_iter(),
+                .map_err(error::PackWriter::StoreOpen)?,
-            odb_opts,
+        );
-        )?);
         let thickener = thickener.to_handle_arc();
         Ok(pack::Bundle::write_to_directory(
             pack,

crates/radicle-node/src/worker/channels.rs

@@ -1,4 +1,3 @@
-use std::convert::Infallible;
 use std::io::{Read, Write};
 use std::ops::Deref;
 use std::{fmt, io, time};
@@ -75,10 +74,9 @@ impl ChannelsFlush {
 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> {
+    fn open(&mut self) -> (&mut Self::Read, &mut Self::Write) {
-        Ok((&mut self.receiver, &mut self.sender))
+        (&mut self.receiver, &mut self.sender)
     }
 }
 
@@ -260,8 +258,6 @@ pub struct ChannelFlushWriter<T = Vec<u8>> {
 }
 
 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()