Improve Uploading Experience — heartwood

The main goal of the patch is to improve the upload experience of repositories that are larger in size, e.g. 300MB and 3GB.

This is achieved by a series of changes, outlined in each commit, and summarised here:

Dynamic timeouts for fetching by passing the timeout through the use Io::Fetch and Control::Open. This removes the DEFAULT_CHANNEL_TIMEOUT and we now default to FETCH_TIMEOUT when it’s not specified.
The Emitter is moved to the radicle crate for re-use in radicle-fetch and radicle-node.
Introduce events for the upload-pack process.
Change the call method for the socket stream to use crossbeam-channel so that events can be subscribed to indefinitely but received with a timeout.
The default value of FETCH_TIMEOUT is increased to 1 hour to allow the larger repositories to finish announcing and being cloned by seeds when initialised.
Improve rad init to report the upload-pack events.

  "crossbeam-channel",
  "cyphernet",
  "fastrand",
  "gix-protocol",
  "io-reactor",
  "lexopt",
  "libc",

 Your Repository ID (RID) is rad:z3Rry7rpdWuGpfjPYGzdJKQADsoNW.
 You can show it any time by running `rad .` from this directory.
 ✓ Repository successfully synced to z6MknSLrJoTcukLrE435hVNQT4JUhbvWLX4kUzqkEStBU8Vi
 ✓ Repository successfully synced to 1 node(s).
 Your repository has been synced to the network and is now discoverable by peers.

 #![allow(clippy::collapsible_else_if)]
 use std::collections::HashSet;
 use std::convert::TryFrom;
 use std::env;
 use std::ffi::OsString;
 use std::path::PathBuf;
 use std::str::FromStr;
 use std::{env, time};
 use anyhow::{anyhow, bail, Context as _};
 use serde_json as json;

 use radicle::explorer::ExplorerUrl;
 use radicle::git::RefString;
 use radicle::identity::{RepoId, Visibility};
 use radicle::node::events::UploadPack;
 use radicle::node::policy::Scope;
 use radicle::node::{Event, Handle, NodeId};
 use radicle::node::{Event, Handle, NodeId, DEFAULT_SUBSCRIBE_TIMEOUT};
 use radicle::prelude::Doc;
 use radicle::{profile, Node};

         return Ok(SyncResult::NodeStopped);
     }
     let mut spinner = term::spinner("Updating inventory..");
     let events = node.subscribe(time::Duration::from_secs(3))?;
     // N.b. indefinitely subscribe to events and set a lower timeout on events
     // below.
     let events = node.subscribe(DEFAULT_SUBSCRIBE_TIMEOUT)?;
     let sessions = node.sessions()?;
     node.update_inventory(rid)?;

     spinner.message("Syncing..");
     let mut replicas = HashSet::new();
     for e in events {
         match e {
             Ok(Event::RefsSynced {
                 remote, rid: rid_, ..
             }) if rid == rid_ => {
                 term::success!("Repository successfully synced to {remote}");
                 replicas.insert(remote);
                 // If we manage to replicate to one of our preferred seeds, we can stop waiting.
                 if config.preferred_seeds.iter().any(|s| s.id == remote) {
                     break;
                 }
             }
             Ok(Event::UploadPack(UploadPack::Write {
                 rid: rid_,
                 remote,
                 progress,
             })) if rid == rid_ => {
                 spinner.message(format!("Uploading {rid} to {remote} {progress}"));
             }
             Ok(Event::UploadPack(UploadPack::Done {
                 rid: rid_,
                 remote,
                 status,
             })) if rid == rid_ => {
                 spinner.message(format!("Upload done for {rid} to {remote}: {status}"));
             }
             Ok(Event::UploadPack(UploadPack::Error {
                 rid: rid_,
                 remote,
                 err,
             })) if rid == rid_ => {
                 spinner.message(format!("Upload error for {rid} to {remote}: {err}"));
             }
             Ok(_) => {
                 // Some other irrelevant event received.
             }

 use std::time;
 use radicle::node::Handle;
 use radicle::node::{Event, Handle};
 pub fn run(node: impl Handle, count: usize, timeout: time::Duration) -> anyhow::Result<()> {
 pub fn run<H>(node: H, count: usize, timeout: time::Duration) -> anyhow::Result<()>
 where
     H: Handle<Event = Result<Event, <H as Handle>::Error>>,
 {
     let events = node.subscribe(timeout)?;
     for (i, event) in events.enumerate() {
     for (i, event) in events.into_iter().enumerate() {
         let event = event?;
         let obj = serde_json::to_string(&event)?;

 use std::{
     borrow::Cow,
     io::{self, BufRead},
     path::PathBuf,
     sync::{atomic::AtomicBool, Arc},
 };
 use std::borrow::Cow;
 use std::io;
 use std::io::BufRead;
 use std::path::PathBuf;
 use std::sync::{atomic::AtomicBool, Arc};
 use gix_features::progress::NestedProgress;
 use gix_pack as pack;
 use gix_protocol::{
     fetch::{self, Delegate, DelegateBlocking},
     handshake::{self, Ref},
     ls_refs, FetchConnection,
 };
 use gix_transport::{
     bstr::BString,
     client,
     client::{ExtendedBufRead, MessageKind},
     Protocol,
 };
 use gix_protocol::fetch;
 use gix_protocol::fetch::{Delegate, DelegateBlocking};
 use gix_protocol::handshake;
 use gix_protocol::handshake::Ref;
 use gix_protocol::ls_refs;
 use gix_protocol::FetchConnection;
 use gix_transport::bstr::BString;
 use gix_transport::client;
 use gix_transport::client::{ExtendedBufRead, MessageKind};
 use gix_transport::Protocol;
 use super::{agent_name, indicate_end_of_interaction, Connection, WantsHaves};

             if !sideband_all {
                 setup_remote_progress(progress, &mut reader);
             }
             let timer = std::time::Instant::now();
             // TODO: remove delegate in favor of functional style to fix progress-hack,
             //       needed as it needs `'static`. As the top-level seems to pass `Discard`,
             //       there should be no repercussions right now.

                 &[],
                 &response,
             )?;
             log::trace!(target: "fetch", "Received pack ({}ms)", timer.elapsed().as_millis());
             assert_eq!(
                 reader.stopped_at(),
                 None,

     if matches!(protocol, Protocol::V2)
         && matches!(conn.mode, FetchConnection::TerminateOnSuccessfulCompletion)
     {
         log::trace!(target: "fetch", "Indicating end of interaction");
         indicate_end_of_interaction(&mut conn)?;
     }

 crossbeam-channel = { version = "0.5.6" }
 cyphernet = { version = "0.5.0", features = ["tor", "dns", "ed25519", "p2p-ed25519"] }
 fastrand = { version = "2.0.0" }
 gix-protocol = { version = "0.41.1", features = ["blocking-client"] }
 io-reactor = { version = "0.5.1", features = ["popol"] }
 lexopt = { version = "0.3.0" }
 libc = { version = "0.2.137" }

 }
 /// Listen for commands on the control socket, and process them.
 pub fn listen<H: Handle<Error = runtime::HandleError> + 'static>(
     listener: UnixListener,
     handle: H,
 ) -> Result<(), Error>
 pub fn listen<E, H>(listener: UnixListener, handle: H) -> Result<(), Error>
 where
     H: Handle<Error = runtime::HandleError> + 'static,
     H::Sessions: serde::Serialize,
     CommandResult<E>: From<H::Event>,
     E: serde::Serialize,
 {
     log::debug!(target: "control", "Control thread listening on socket..");
     let nid = handle.nid()?;

     Io(#[from] io::Error),
 }
 fn command<H: Handle<Error = runtime::HandleError> + 'static>(
     stream: &UnixStream,
     mut handle: H,
 ) -> Result<(), CommandError>
 fn command<E, H>(stream: &UnixStream, mut handle: H) -> Result<(), CommandError>
 where
     H: Handle<Error = runtime::HandleError> + 'static,
     H::Sessions: serde::Serialize,
     CommandResult<E>: From<H::Event>,
     E: serde::Serialize,
 {
     let mut reader = BufReader::new(stream);
     let mut writer = LineWriter::new(stream);

         Command::Subscribe => match handle.subscribe(MAX_TIMEOUT) {
             Ok(events) => {
                 for e in events {
                     let event = e?;
                     CommandResult::Okay(event).to_writer(&mut writer)?;
                     CommandResult::from(e).to_writer(&mut writer)?;
                 }
             }
             Err(e) => return Err(CommandError::Runtime(e)),

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

 pub use handle::Error as HandleError;
 pub use handle::Handle;
 pub use node::events::Emitter;
 /// A client error.
 #[derive(Error, Debug)]

     GitVersion(#[from] git::VersionError),
 }
 /// Publishes events to subscribers.
 #[derive(Debug, Clone)]
 pub struct Emitter<T> {
     subscribers: Arc<Mutex<Vec<chan::Sender<T>>>>,
 }
 impl<T> Default for Emitter<T> {
     fn default() -> Emitter<T> {
         Emitter {
             subscribers: Default::default(),
         }
     }
 }
 impl<T: Clone> Emitter<T> {
     /// Emit event to subscribers and drop those who can't receive it.
     pub(crate) fn emit(&self, event: T) {
         self.subscribers
             .lock()
             .unwrap()
             .retain(|s| s.try_send(event.clone()).is_ok());
     }
     /// Subscribe to events stream.
     pub fn subscribe(&self) -> chan::Receiver<T> {
         let (sender, receiver) = chan::unbounded();
         let mut subs = self.subscribers.lock().unwrap();
         subs.push(sender);
         receiver
     }
 }
 /// Holds join handles to the client threads, as well as a client handle.
 pub struct Runtime {
     pub id: NodeId,

 impl radicle::node::Handle for Handle {
     type Sessions = Vec<radicle::node::Session>;
     type Events = Events;
     type Event = Event;
     type Error = Error;
     fn nid(&self) -> Result<NodeId, Self::Error> {

         receiver.recv().map_err(Error::from)
     }
     fn subscribe(
         &self,
         _timeout: time::Duration,
     ) -> Result<Box<dyn Iterator<Item = Result<Event, Error>>>, Error> {
         Ok(Box::new(self.events().into_iter().map(Ok)))
     fn subscribe(&self, _timeout: time::Duration) -> Result<Self::Events, Self::Error> {
         Ok(self.events())
     }
     fn sessions(&self) -> Result<Self::Sessions, Error> {

 pub const MAX_RECONNECTION_DELTA: LocalDuration = LocalDuration::from_mins(60);
 /// Connection retry delta used for ephemeral peers that failed to connect previously.
 pub const CONNECTION_RETRY_DELTA: LocalDuration = LocalDuration::from_mins(10);
 /// How long to wait for a fetch to stall before aborting.
 pub const FETCH_TIMEOUT: time::Duration = time::Duration::from_secs(9);
 /// How long to wait for a fetch to stall before aborting, default is 1 hour.
 pub const FETCH_TIMEOUT: time::Duration = time::Duration::from_secs(3600);
 /// Maximum external address limit imposed by message size limits.
 pub use message::ADDRESS_LIMIT;

     from: NodeId,
     /// Refs being fetched.
     refs_at: Vec<RefsAt>,
     /// The timeout given for the fetch request.
     timeout: time::Duration,
     /// Result channel.
     channel: Option<chan::Sender<FetchResult>>,
 }

     pub fn local_time(&self) -> LocalTime {
         self.clock
     }
     pub fn emitter(&self) -> Emitter<Event> {
         self.emitter.clone()
     }
 }
 impl<D, S, G> Service<D, S, G>

                         rid,
                         refs_at,
                         from,
                         timeout,
                         channel,
                     };
                     if self.queue.contains(&fetch) {

                     .expect("Service::dequeue_fetch: error accessing repo seeding configuration");
                 // Keep dequeueing if there was nothing to fetch, otherwise break.
                 if self.fetch_refs_at(rid, from, refs, repo_entry.scope, FETCH_TIMEOUT, channel) {
                 if self.fetch_refs_at(rid, from, refs, repo_entry.scope, timeout, channel) {
                     break;
                 }
             } else {
                 // If no refs are specified, always do a full fetch.
                 self.fetch(rid, from, FETCH_TIMEOUT, channel);
                 self.fetch(rid, from, timeout, channel);
                 break;
             }
         }

 impl radicle::node::Handle for Handle {
     type Error = HandleError;
     type Sessions = Vec<radicle::node::Session>;
     type Events = Vec<Self::Event>;
     type Event = Result<Event, Self::Error>;
     fn nid(&self) -> Result<NodeId, Self::Error> {
         Ok(NodeId::from_str("z6MkhaXgBZDvotDkL5257faiztiGiC2QtKLGpbnnEGta2doK").unwrap())

         Ok(self.following.lock().unwrap().insert(id))
     }
     fn subscribe(
         &self,
         _timeout: time::Duration,
     ) -> Result<Box<dyn Iterator<Item = Result<Event, Self::Error>>>, Self::Error> {
         Ok(Box::new(std::iter::empty()))
     fn subscribe(&self, _timeout: time::Duration) -> Result<Self::Events, Self::Error> {
         Ok(vec![])
     }
     fn unfollow(&mut self, id: NodeId) -> Result<bool, Self::Error> {

 use crate::prelude::Deserializer;
 use crate::service;
 use crate::service::io::Io;
 use crate::service::FETCH_TIMEOUT;
 use crate::service::{session, DisconnectReason, Service};
 use crate::wire::frame;
 use crate::wire::frame::{Frame, FrameData, StreamId};

     responder: cyphernet::encrypt::noise::OneWayPattern::Known,
 };
 /// 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(30);
 /// Default time to wait until a network connection is considered inactive.
 pub const DEFAULT_CONNECTION_TIMEOUT: time::Duration = time::Duration::from_secs(6);

     }
     /// Open a new stream.
     fn open(&mut self) -> (StreamId, worker::Channels) {
     fn open(&mut self, timeout: time::Duration) -> (StreamId, worker::Channels) {
         self.seq += 1;
         let id = StreamId::git(self.link)
             .nth(self.seq)
             .expect("Streams::open: too many streams");
         let channels = self
             .register(id)
             .register(id, timeout)
             .expect("Streams::open: stream was already open");
         (id, channels)
     }
     /// Register an open stream.
     fn register(&mut self, stream: StreamId) -> Option<worker::Channels> {
         let (wire, worker) = worker::Channels::pair(DEFAULT_CHANNEL_TIMEOUT)
     fn register(&mut self, stream: StreamId, timeout: time::Duration) -> Option<worker::Channels> {
         let (wire, worker) = worker::Channels::pair(timeout)
             .expect("Streams::register: fatal: unable to create channels");
         match self.streams.entry(stream) {

                             })) => {
                                 log::debug!(target: "wire", "Received `open` command for stream {stream} from {nid}");
                                 let Some(channels) = streams.register(stream) else {
                                 let Some(channels) = streams.register(stream, FETCH_TIMEOUT) else {
                                     log::warn!(target: "wire", "Peer attempted to open already-open stream stream {stream}");
                                     continue;
                                 };
                                 let task = Task {
                                     fetch: FetchRequest::Responder { remote: *nid },
                                     fetch: FetchRequest::Responder {
                                         remote: *nid,
                                         emitter: self.service.emitter(),
                                     },
                                     stream,
                                     channels,
                                 };

                         log::error!(target: "wire", "Peer {remote} is not connected: dropping fetch");
                         continue;
                     };
                     let (stream, channels) = streams.open();
                     let (stream, channels) = streams.open(timeout);
                     log::debug!(target: "wire", "Opened new stream with id {stream} for {rid} and remote {remote}");

 pub mod fetch;
 pub mod garbage;
 use std::io;
 use std::path::PathBuf;
 use std::{io, time};
 use crossbeam_channel as chan;
 use radicle::identity::RepoId;
 use radicle::node::notifications;
 use radicle::node::{notifications, Event};
 use radicle::prelude::NodeId;
 use radicle::storage::refs::RefsAt;
 use radicle::storage::{ReadRepository, ReadStorage};
 use radicle::{cob, crypto, Storage};
 use radicle_fetch::FetchLimit;
 use crate::runtime::{thread, Handle};
 use crate::runtime::{thread, Emitter, Handle};
 use crate::service::policy;
 use crate::service::policy::Policy;
 use crate::wire::StreamId;

         remote: NodeId,
         /// If this fetch is for a particular set of `rad/sigrefs`.
         refs_at: Option<Vec<RefsAt>>,
         /// Fetch timeout.
         timeout: time::Duration,
     },
     /// Server is responding to a fetch request by uploading the
     /// specified `refspecs` sent by the client.
     Responder {
         /// Remote peer we are interacting with.
         remote: NodeId,
         /// Reporter for upload-pack progress.
         emitter: Emitter<Event>,
     },
 }
 impl FetchRequest {
     pub fn remote(&self) -> NodeId {
         match self {
             Self::Initiator { remote, .. } | Self::Responder { remote } => *remote,
             Self::Initiator { remote, .. } | Self::Responder { remote, .. } => *remote,
         }
     }
 }

                 rid,
                 remote,
                 refs_at,
                 // TODO: nowhere to use this currently
                 timeout: _timeout,
             } => {
                 log::debug!(target: "worker", "Worker processing outgoing fetch for {rid}");
                 let result = self.fetch(rid, remote, refs_at, channels, notifs);
                 FetchResult::Initiator { rid, result }
             }
             FetchRequest::Responder { remote } => {
             FetchRequest::Responder { remote, emitter } => {
                 log::debug!(target: "worker", "Worker processing incoming fetch for {remote} on stream {stream}..");
                 let (mut stream_r, stream_w) = channels.split();