fetch: improve the fetch protocol — heartwood

Logging the stages of the fetch protocol is added to get some insight of how far a fetch gets, while also providing some simple timings using Instant::elapsed.

This allowed me to run a local client version of this fetch protocol and see what was happening. The 9s timeout was hit for the channel methods. So this was increased to 30s.

From there, I was able to see that a rad clone of an offending repository was actually “finishing” but the final error would say that the other side had disconnected. This lead to the realisation that the call to done can be a fire and forget – log the error and continue on with validation and refdb updates. I was then able to confirm that the clone of the offending repository was able to be performed.

 mod stage;
 mod state;
 use gix_protocol::handshake;
 pub use handle::Handle;
 pub use policy::{Allowed, BlockList, Scope};
 use radicle::storage::ReadRepository as _;
 pub use state::{FetchLimit, FetchResult};
 pub use transport::Transport;
 use std::io;
 use std::time::Instant;
 use radicle::crypto::PublicKey;
 use radicle::storage::refs::RefsAt;

 where
     S: transport::ConnectionStream,
 {
     let start = Instant::now();
     let local = *handle.local();
     if local == remote {
         return Err(Error::ReplicateSelf);
     }
     let handshake = handle
         .transport
         .handshake()
         .map_err(|err| Error::Handshake { err })?;
     let handshake = perform_handshake(handle)?;
     let state = FetchState::default();
     // N.b. ensure that we ignore the local peer's key.
     handle.blocked.extend([local]);
     state
     let result = state
         .run(handle, &handshake, limit, remote, refs_at)
         .map_err(Error::Protocol)
         .map_err(Error::Protocol);
     log::debug!(
         target: "fetch",
         "finished pull of {} ({}ms)",
         handle.repo.id(),
         start.elapsed().as_millis()
     );
     result
 }
 /// Clone changes from the `remote`.

 where
     S: transport::ConnectionStream,
 {
     let start = Instant::now();
     if *handle.local() == remote {
         return Err(Error::ReplicateSelf);
     }
     let handshake = handle
         .transport
         .handshake()
         .map_err(|err| Error::Handshake { err })?;
     let handshake = perform_handshake(handle)?;
     let state = FetchState::default();
     state
     let result = state
         .run(handle, &handshake, limit, remote, None)
         .map_err(Error::Protocol)
         .map_err(Error::Protocol);
     log::debug!(
         target: "fetch",
         "finished clone of {} ({}ms)",
         handle.repo.id(),
         start.elapsed().as_millis(),
     );
     result
 }
 fn perform_handshake<S>(handle: &mut Handle<S>) -> Result<handshake::Outcome, Error>
 where
     S: transport::ConnectionStream,
 {
     handle.transport.handshake().map_err(|err| {
         log::warn!(target: "fetch", "failed to perform handshake: {err}");
         Error::Handshake { err }
     })
 }

 use std::collections::{BTreeMap, BTreeSet};
 use std::time::Instant;
 use gix_protocol::handshake;
 use radicle::crypto::PublicKey;

     where
         S: transport::ConnectionStream,
     {
         let start = Instant::now();
         // N.b. we always fetch the `rad/id` since our delegate set
         // might be further ahead than theirs, e.g. we are the
         // deciding vote on adding a delegate.

                 limit: limit.special,
             },
         )?;
         log::debug!(target: "fetch", "fetched rad/id ({}ms)", start.elapsed().as_millis());
         // N.b. The error case here should not happen. In the case of
         // a `clone` we have asked for refs/rad/id and ensured it was

             remote,
             refs_at,
         )?;
         log::debug!(
             target: "fetch",
             "fetched rad/sigrefs for {} remotes ({}ms)",
             signed_refs.len(),
             start.elapsed().as_millis()
         );
         let data_refs = stage::DataRefs {
             remote,

             limit: limit.refs,
         };
         self.run_stage(handle, handshake, &data_refs)?;
         log::debug!(
             target: "fetch",
             "fetched data refs for {} remotes ({}ms)",
             data_refs.remotes.len(),
             start.elapsed().as_millis()
         );
         // N.b. signal to exit the upload-pack sequence
         // We're finished fetching on this side, and all that's left
         // is validation.
         match handle.transport.done() {
             Ok(()) => log::debug!(target: "fetch", "sent done signal to remote"),
             Err(err) => {
                 log::warn!(target: "fetch", "attempted to send done to remote {remote}: {err}")
             }
         }
         // Run validation of signed refs, pruning any offending
         // remotes from the tips, thus not updating the production Git

                     .into_values()
                     .flat_map(|ups| ups.into_iter()),
             )?;
             log::debug!(target: "fetch", "applied updates ({}ms)", start.elapsed().as_millis());
             Ok(FetchResult::Success {
                 applied,
                 remotes,
                 warnings,
             })
         } else {
             log::debug!(
                 target: "fetch",
                 "fetch failed {} warnings and {} failures ({}ms)",
                 warnings.len(),
                 failures.len(),
                 start.elapsed().as_millis()
             );
             Ok(FetchResult::Failed { warnings, failures })
         }
     }

 use std::os::unix::net::UnixListener;
 use std::path::PathBuf;
 use std::sync::{Arc, Mutex};
 use std::{fs, io, net, time};
 use std::{fs, io, net};
 use crossbeam_channel as chan;
 use cyphernet::Ecdh;

             cobs_cache,
             worker::Config {
                 capacity: 8,
                 timeout: time::Duration::from_secs(9),
                 storage: storage.clone(),
                 fetch,
                 policy,

             match self.db.addresses().get(announcer) {
                 Ok(node) => {
                     if node.is_none() {
                         debug!(target: "service", "Ignoring announcement from unknown node {announcer}");
                         trace!(target: "service", "Ignoring announcement from unknown node {announcer}");
                         return Ok(false);
                     }
                 }

             trace!(target: "service", "Rate limiting message from {remote} ({})", peer.addr);
             return Ok(());
         }
         message.log(log::Level::Debug, remote, Link::Inbound);
         message.log(log::Level::Trace, remote, Link::Inbound);
         trace!(target: "service", "Received message {:?} from {}", &message, peer.id);

     }
     pub fn write(&mut self, remote: &Session, msg: Message) {
         msg.log(log::Level::Debug, &remote.id, Link::Outbound);
         msg.log(log::Level::Trace, &remote.id, Link::Outbound);
         trace!(target: "service", "Write {:?} to {}", &msg, remote);
         self.io.push_back(Io::Write(remote.id, vec![msg]));

 /// Default time to wait to receive something from a worker channel. Applies to
 /// workers waiting for data from remotes as well.
 pub const DEFAULT_CHANNEL_TIMEOUT: time::Duration = time::Duration::from_secs(9);
 pub const DEFAULT_CHANNEL_TIMEOUT: time::Duration = time::Duration::from_secs(30);
 /// Default time to wait until a network connection is considered inactive.
 pub const DEFAULT_CONNECTION_TIMEOUT: time::Duration = time::Duration::from_secs(30);

 pub struct Config {
     /// Number of worker threads.
     pub capacity: usize,
     /// Timeout for all operations.
     pub timeout: time::Duration,
     /// Git storage.
     pub storage: Storage,
     /// Configuration for performing fetched.

 /// Default radicle protocol port.
 pub const DEFAULT_PORT: u16 = 8776;
 /// Default timeout when waiting for the node to respond with data.
 pub const DEFAULT_TIMEOUT: time::Duration = time::Duration::from_secs(9);
 pub const DEFAULT_TIMEOUT: time::Duration = time::Duration::from_secs(30);
 /// Maximum length in bytes of a node alias.
 pub const MAX_ALIAS_LENGTH: usize = 32;
 /// Penalty threshold at which point we avoid connecting to this node.