node: reduce default upload-pack timeout

From empirical testing, the `FETCH_TIMEOUT` can be set to a lower value of 3s.
For a repository that is about 300MiB, the process will successfully send all
packfile bytes and the references will be written.

The `--timeout` option is also used in the `git-upload-pack` process for good
measure.

Signed-off-by: Fintan Halpenny <fintan.halpenny@gmail.com>
X-Clacks-Overhead: GNU Terry Pratchett

radicle-node/src/service.rs

@@ -98,8 +98,8 @@ pub const MIN_RECONNECTION_DELTA: LocalDuration = LocalDuration::from_secs(3);
 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, default is 1 hour.
+/// How long to wait for a fetch to stall before aborting, default is 3s.
-pub const FETCH_TIMEOUT: time::Duration = time::Duration::from_secs(3600);
+pub const FETCH_TIMEOUT: time::Duration = time::Duration::from_secs(3);
 
 /// Maximum external address limit imposed by message size limits.
 pub use message::ADDRESS_LIMIT;

radicle-node/src/worker.rs

@@ -238,6 +238,7 @@ impl Worker {
             FetchRequest::Responder { remote, emitter } => {
                 log::debug!(target: "worker", "Worker processing incoming fetch for {remote} on stream {stream}..");
 
+                let timeout = channels.timeout();
                 let (mut stream_r, stream_w) = channels.split();
                 let header = match upload_pack::pktline::git_request(&mut stream_r) {
                     Ok(header) => header,
@@ -265,6 +266,7 @@ impl Worker {
                     &header,
                     stream_r,
                     stream_w,
+                    timeout,
                 )
                 .map(|_| ())
                 .map_err(|e| e.into());

radicle-node/src/worker/channels.rs

@@ -34,6 +34,10 @@ impl ChannelsFlush {
     pub fn split(&mut self) -> (&mut ChannelReader, &mut ChannelFlushWriter) {
         (&mut self.receiver, &mut self.sender)
     }
+
+    pub fn timeout(&self) -> time::Duration {
+        self.sender.writer.timeout.max(self.receiver.timeout)
+    }
 }
 
 impl radicle_fetch::transport::ConnectionStream for ChannelsFlush {

radicle-node/src/worker/upload_pack.rs

@@ -1,7 +1,7 @@
 use std::io;
 use std::io::Write;
 use std::process::{Command, ExitStatus, Stdio};
-use std::time::Instant;
+use std::time::{Duration, Instant};
 
 use gix_protocol::transport::bstr::ByteSlice;
 use radicle::identity::RepoId;
@@ -27,6 +27,7 @@ pub fn upload_pack<R, W>(
     header: &pktline::GitRequest,
     mut recv: R,
     send: W,
+    timeout: Duration,
 ) -> io::Result<ExitStatus>
 where
     R: io::Read + Send,
@@ -72,6 +73,7 @@ where
                 "lsrefs.unborn=ignore",
                 "upload-pack",
                 "--strict",
+                format!("--timeout={}", timeout.as_secs()).as_str(),
                 ".",
             ])
             .stdout(Stdio::piped())
@@ -113,6 +115,12 @@ where
                         log::debug!(target: "worker", "Exiting upload-pack reader thread for {}", header.repo);
                         break;
                     }
+                    // N.b. if the read timed out, ensure that the sender isn't
+                    // still sending messages.
+                    Err(e) if e.kind() == io::ErrorKind::TimedOut => {
+                        log::warn!(target: "worker", "Read channel timed out for upload-pack {}", header.repo);
+                        break;
+                    }
                     Err(e) => {
                         log::error!(target: "worker", "Error on upload-pack channel read for {}: {e}", header.repo);
                         emitter.emit(events::UploadPack::error(header.repo, remote, e).into());
@@ -167,7 +175,7 @@ impl<W> Reporter<W> {
             }
         };
         log::trace!(target: "worker", "upload-pack progress: {event:?}");
-        self.emitter.emit(event.into())
+        self.emitter.emit(event.into());
     }
 
     fn as_upload_pack_progress(buf: &[u8]) -> Option<events::upload_pack::Progress> {