node: Fix bug in upload-pack

We were not properly exiting one of the threads that was copying
data from the stream to the upload-pack process. This is a fix.

radicle-node/src/service.rs

@@ -27,7 +27,7 @@ use crate::crypto;
 use crate::crypto::{Signer, Verified};
 use crate::identity::{Doc, Id};
 use crate::node;
-use crate::node::{Address, Features, FetchLookup, FetchResult};
+use crate::node::{Address, Features, FetchError, FetchLookup, FetchResult};
 use crate::prelude::*;
 use crate::service::message::{Announcement, AnnouncementMessage, Ping};
 use crate::service::message::{NodeAnnouncement, RefsAnnouncement};
@@ -499,22 +499,39 @@ where
     pub fn fetched(&mut self, result: FetchResult) {
         let remote = result.remote;
         let rid = result.rid;
-
+        let namespaces = result.namespaces;
-        match &result.result {
+        let result = match result.result {
             Ok(updated) => {
                 self.reactor.event(Event::RefsFetched {
                     remote,
                     rid,
                     updated: updated.clone(),
                 });
+                Ok(updated)
             }
             Err(err) => {
-                error!("Fetch failed for {rid} from {remote}: {err}");
+                error!(target: "service", "Fetch failed for {rid} from {remote}: {err}");
+
+                if let FetchError::Io(_) = err {
+                    self.reactor
+                        .disconnect(result.remote, DisconnectReason::Fetch(err));
+                    return;
+                } else {
+                    Err(err)
                 }
             }
+        };
 
         if let Some(results) = self.fetch_reqs.get(&rid) {
-            if results.send(result).is_err() {
+            if results
+                .send(FetchResult {
+                    rid,
+                    remote,
+                    namespaces,
+                    result,
+                })
+                .is_err()
+            {
                 self.fetch_reqs.remove(&rid);
             }
         }
@@ -1151,6 +1168,8 @@ pub enum DisconnectReason {
     /// Error with an underlying established connection. Sometimes, reconnecting
     /// after such an error is possible.
     Connection(Arc<dyn std::error::Error + Sync + Send>),
+    /// Error with a fetch.
+    Fetch(FetchError),
     /// Session error.
     Session(session::Error),
 }
@@ -1169,6 +1188,7 @@ impl DisconnectReason {
             Self::Dial(_) => false,
             Self::Connection(_) => true,
             Self::Session(..) => false,
+            Self::Fetch(_) => true,
         }
     }
 }
@@ -1179,6 +1199,7 @@ impl fmt::Display for DisconnectReason {
             Self::Dial(err) => write!(f, "{}", err),
             Self::Connection(err) => write!(f, "{}", err),
             Self::Session(err) => write!(f, "error: {}", err),
+            Self::Fetch(err) => write!(f, "fetch: {}", err),
         }
     }
 }

radicle-node/src/service/reactor.rs

@@ -63,7 +63,7 @@ impl Reactor {
     }
 
     pub fn write(&mut self, remote: NodeId, msg: Message) {
-        debug!("Write {:?} to {}", &msg, remote);
+        debug!(target: "service", "Write {:?} to {}", &msg, remote);
 
         self.io.push_back(Io::Write(remote, vec![msg]));
     }
@@ -72,6 +72,7 @@ impl Reactor {
         let msgs = msgs.into_iter().collect::<Vec<_>>();
         for (ix, msg) in msgs.iter().enumerate() {
             debug!(
+                target: "service",
                 "Write {:?} message to {} ({}/{})",
                 msg,
                 remote,

radicle-node/src/worker.rs

@@ -1,6 +1,6 @@
 use std::io::prelude::*;
 use std::thread::JoinHandle;
-use std::{env, io, net, process, str, thread, time};
+use std::{env, io, net, process, thread, time};
 
 use crossbeam_channel as chan;
 use cyphernet::EcSign;
@@ -125,9 +125,9 @@ impl<G: Signer + EcSign + 'static> Worker<G> {
         let tunnel_addr = tunnel.local_addr()?;
         let mut cmd = process::Command::new("git");
         cmd.current_dir(repo.path())
-            .env("GIT_PROTOCOL", "2")
             .env_clear()
             .envs(env::vars().filter(|(k, _)| k == "PATH" || k.starts_with("GIT_TRACE")))
+            .args(["-c", "protocol.version=2"])
             .arg("fetch")
             .arg("--atomic")
             .arg("--verbose")
@@ -178,7 +178,7 @@ impl<G: Signer + EcSign + 'static> Worker<G> {
             .current_dir(repo.path())
             .env_clear()
             .envs(env::vars().filter(|(k, _)| k == "PATH" || k.starts_with("GIT_TRACE")))
-            .env("GIT_PROTOCOL", "2")
+            .args(["-c", "protocol.version=2"])
             .arg("upload-pack")
             .arg("--strict") // The path to the git repo must be exact.
             .arg(".")
@@ -190,10 +190,10 @@ impl<G: Signer + EcSign + 'static> Worker<G> {
         let mut stdin = child.stdin.take().unwrap();
         let mut stdout = child.stdout.take().unwrap();
         let mut stderr = child.stderr.take().unwrap();
-        let mut reader = GitReader::new(drain, stream_r);
+        let mut reader = pktline::GitReader::new(drain, stream_r);
 
         match reader.read_command_pkt_line() {
-            Ok(cmd) => {
+            Ok((cmd, _pktline)) => {
                 log::debug!(
                     target: "worker",
                     "Parsed git command packet-line for {}: {:?}", fetch.repo, cmd
@@ -204,27 +204,34 @@ impl<G: Signer + EcSign + 'static> Worker<G> {
                     )));
                 }
             }
-            Err(_) => {
+            Err(err) => {
-                return Err(FetchError::Git(git::raw::Error::from_str(
+                return Err(FetchError::Git(git::raw::Error::from_str(&format!(
-                    "error parsing git command packet-line",
+                    "error parsing git command packet-line: {err}"
-                )));
+                ))));
             }
         }
 
-        thread::scope(|scope| {
+        thread::scope::<_, Result<Vec<RefUpdate>, FetchError>>(|scope| {
+            // Output of `upload-pack` is sent back to the remote peer.
+            let outgoing = scope.spawn(move || io::copy(&mut stdout, stream_w));
+
+            let mut buf = [0; 65536];
             // Data coming from the remote peer is written to the standard input of the
             // `upload-pack` process.
-            // FIXME: This sometimes returns a `WouldBlock`.
+            while !outgoing.is_finished() {
-            let t = scope.spawn(move || io::copy(&mut reader, &mut stdin));
+                let n = reader.read_pkt_line(&mut buf)?;
-            // Output of `upload-pack` is sent back to the remote peer.
+
-            io::copy(&mut stdout, stream_w)?;
+                stdin.write_all(&buf[..n]).unwrap();
+                log::trace!(target: "worker", "Received {:?}", String::from_utf8_lossy(&buf[..n]));
+
+                if &buf[..n] == pktline::DONE {
+                    break;
+                }
+            }
             // SAFETY: The thread should not panic, but if it does, we bubble up the panic.
-            t.join().unwrap()?;
+            outgoing.join().unwrap()?;
 
-            Ok::<_, FetchError>(())
-        })?;
             let status = child.wait()?;
-
             if let Some(status) = status.code() {
                 log::debug!(target: "worker", "Upload-pack for {} exited with status {:?}", fetch.repo, status);
             } else {
@@ -238,8 +245,8 @@ impl<G: Signer + EcSign + 'static> Worker<G> {
                 let err = String::from_utf8_lossy(&err);
                 log::debug!(target: "worker", "Upload-pack for {}: stderr: {}", fetch.repo, err);
             }
-
             Ok(vec![])
+        })
     }
 }
 
@@ -291,13 +298,26 @@ impl WorkerPool {
     }
 }
 
-pub struct GitReader<'a, R> {
+mod pktline {
+    use std::io;
+    use std::io::Read;
+    use std::str;
+
+    use super::Id;
+
+    pub const HEADER_LEN: usize = 4;
+    pub const FLUSH_PKT: &[u8; HEADER_LEN] = b"0000";
+    pub const DELIM_PKT: &[u8; HEADER_LEN] = b"0001";
+    pub const RESPONSE_END_PKT: &[u8; HEADER_LEN] = b"0002";
+    pub const DONE: &[u8] = b"0009done\n";
+
+    pub struct GitReader<'a, R> {
         drain: Vec<u8>,
         stream: &'a mut R,
-}
+    }
 
-impl<'a, R: io::Read> GitReader<'a, R> {
+    impl<'a, R: io::Read> GitReader<'a, R> {
-    fn new(drain: Vec<u8>, stream: &'a mut R) -> Self {
+        pub fn new(drain: Vec<u8>, stream: &'a mut R) -> Self {
             Self { drain, stream }
         }
 
@@ -305,33 +325,37 @@ impl<'a, R: io::Read> GitReader<'a, R> {
         ///
         /// Example: `0032git-upload-pack /project.git\0host=myserver.com\0`
         ///
-    fn read_command_pkt_line(&mut self) -> io::Result<GitCommand> {
+        pub fn read_command_pkt_line(&mut self) -> io::Result<(GitCommand, Vec<u8>)> {
             let mut pktline = [0u8; 1024];
             let length = self.read_pkt_line(&mut pktline)?;
-        let Some(cmd) = GitCommand::parse(&pktline[..length]) else {
+            let Some(cmd) = GitCommand::parse(&pktline[4..length]) else {
             return Err(io::ErrorKind::InvalidInput.into());
         };
-        Ok(cmd)
+            Ok((cmd, Vec::from(&pktline[..length])))
         }
 
         /// Parse a Git packet-line.
-    fn read_pkt_line(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+        pub fn read_pkt_line(&mut self, buf: &mut [u8]) -> io::Result<usize> {
-        let mut length = [0; 4];
+            self.read_exact(&mut buf[..HEADER_LEN])?;
-        self.read_exact(&mut length)?;
 
-        let length = str::from_utf8(&length)
+            if &buf[..HEADER_LEN] == FLUSH_PKT
+                || &buf[..HEADER_LEN] == DELIM_PKT
+                || &buf[..HEADER_LEN] == RESPONSE_END_PKT
+            {
+                return Ok(HEADER_LEN);
+            }
+            let length = str::from_utf8(&buf[..HEADER_LEN])
                 .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e.to_string()))?;
             let length = usize::from_str_radix(length, 16)
                 .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e.to_string()))?;
-        let remaining = length - 4;
 
-        self.read_exact(&mut buf[..remaining])?;
+            self.read_exact(&mut buf[HEADER_LEN..length])?;
 
-        Ok(remaining)
+            Ok(length)
+        }
     }
-}
 
-impl<'a, R: io::Read> io::Read for GitReader<'a, R> {
+    impl<'a, R: io::Read> io::Read for GitReader<'a, R> {
         fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
             if !self.drain.is_empty() {
                 let count = buf.len().min(self.drain.len());
@@ -342,17 +366,17 @@ impl<'a, R: io::Read> io::Read for GitReader<'a, R> {
             }
             self.stream.read(buf)
         }
-}
+    }
 
-#[derive(Debug)]
+    #[derive(Debug)]
-pub struct GitCommand {
+    pub struct GitCommand {
         pub repo: Id,
         pub path: String,
         pub host: Option<(String, Option<u16>)>,
         pub extra: Vec<(String, Option<String>)>,
-}
+    }
 
-impl GitCommand {
+    impl GitCommand {
         /// Parse a Git command from a packet-line.
         fn parse(input: &[u8]) -> Option<Self> {
             let input = str::from_utf8(input).ok()?;
@@ -390,4 +414,5 @@ impl GitCommand {
                 extra,
             })
         }
+    }
 }