node: Refactor fetch response code

Only user-requested fetches require a response.

Signed-off-by: Alexis Sellier <self@cloudhead.io>

radicle-node/src/control.rs

@@ -13,7 +13,6 @@ use crate::client;
 use crate::identity::Id;
 use crate::node;
 use crate::service::FetchLookup;
-use crate::service::FetchResult;
 
 #[derive(thiserror::Error, Debug)]
 pub enum Error {
@@ -242,19 +241,19 @@ fn fetch<W: Write, H: Handle<Error = client::handle::Error, FetchLookup = FetchL
             )?;
 
             for result in results.iter() {
-                match result {
+                match result.result {
-                    FetchResult::Fetched { from, updated } => {
+                    Ok(updated) => {
-                        writeln!(writer, "ok: {} fetched from {}", &id, from)?;
+                        writeln!(writer, "ok: {} fetched from {}", &id, result.remote)?;
 
                         for update in updated {
                             writeln!(writer, "{}", update)?;
                         }
                     }
-                    FetchResult::Error { from, error } => {
+                    Err(err) => {
                         writeln!(
                             writer,
                             "error: {} failed to fetch from {}: {}",
-                            &id, from, error
+                            &id, result.remote, err
                         )?;
                     }
                 }

radicle-node/src/service.rs

@@ -131,24 +131,10 @@ pub enum FetchLookup {
 /// Result of a fetch request from a specific seed.
 #[derive(Debug)]
 #[allow(clippy::large_enum_variant)]
-pub enum FetchResult {
+pub struct FetchResult {
-    /// Successful fetch from a seed.
+    pub rid: Id,
-    Fetched {
+    pub remote: NodeId,
-        from: NodeId,
+    pub result: Result<Vec<RefUpdate>, FetchError>,
-        updated: Vec<RefUpdate>,
-    },
-    /// Error fetching the resource from a seed.
-    Error { from: NodeId, error: FetchError },
-}
-
-impl FetchResult {
-    /// Get the remote node id.
-    pub fn remote(&self) -> &NodeId {
-        match self {
-            Self::Fetched { from, .. } => from,
-            Self::Error { from, .. } => from,
-        }
-    }
 }
 
 /// Function used to query internal service state.
@@ -226,6 +212,8 @@ pub struct Service<R, A, S, G> {
     rng: Rng,
     /// Whether our local inventory no long represents what we have announced to the network.
     out_of_sync: bool,
+    /// Fetch requests initiated by user, which are waiting for results.
+    fetch_reqs: HashMap<Id, chan::Sender<FetchResult>>,
     /// Current tracked repository bloom filter.
     filter: Filter,
     /// Last time the service was idle.
@@ -289,6 +277,7 @@ where
             reactor: Reactor::default(),
             sessions,
             out_of_sync: false,
+            fetch_reqs: HashMap::new(),
             filter: Filter::empty(),
             last_idle: LocalTime::default(),
             last_sync: LocalTime::default(),
@@ -456,8 +445,14 @@ where
                     return;
                 }
 
-                let Ok(seeds) = self.routing.get(&id) else {
+                let seeds = match self.routing.get(&id) {
-                    todo!();
+                    Ok(seeds) => seeds,
+                    Err(err) => {
+                        log::error!("Error reading routing table for {id}: {err}");
+                        resp.send(FetchLookup::NotFound).ok();
+
+                        return;
+                    }
                 };
                 let Some(seeds) = NonEmpty::from_vec(seeds.into_iter().collect()) else {
                     log::warn!("No seeds found for {}", id);
@@ -474,17 +469,12 @@ where
                 })
                 .ok();
 
+                self.fetch_reqs.insert(id, results_send);
+
                 // TODO: Limit the number of seeds we fetch from? Randomize?
                 for seed in seeds {
-                    let session = self.sessions.get_mut(&seed).unwrap();
+                    if let Err(err) = self.fetch(id, seed) {
-                    if let Some(fetch) = session.fetch(id, results_send.clone()) {
+                        log::error!("Error initiating fetch for {id} from {seed}: {err}");
-                        self.reactor.write(session.id, fetch);
-                        self.reactor
-                            .fetch(session.id, id, Namespaces::default(), true);
-                    } else {
-                        // TODO: If we can't fetch, it's because we're already fetching from
-                        // this peer. So we need to queue the request, or find another peer.
-                        todo!();
                     }
                 }
             }
@@ -525,18 +515,38 @@ where
         }
     }
 
+    pub fn fetch(&mut self, rid: Id, seed: NodeId) -> Result<(), Error> {
+        let session = self.sessions.get_mut(&seed).unwrap();
+        if let Some(fetch) = session.fetch(rid) {
+            self.reactor.write(session.id, fetch);
+            self.reactor
+                .fetch(session.id, rid, Namespaces::default(), true);
+        } else {
+            // TODO: If we can't fetch, it's because we're already fetching from
+            // this peer. So we need to queue the request, or find another peer.
+            todo!();
+        }
+        Ok(())
+    }
+
     pub fn repo_fetched(&mut self, result: FetchResult) {
-        // TODO(cloudhead): handle completed job with service business logic
+        let remote = result.remote;
-        // TODO: Downgrade session to gossip protocol.
+        let rid = result.rid;
-        if let Some(session) = self.sessions.get_mut(result.remote()) {
+
-            if let session::State::Connected { protocol, .. } = &session.state {
+        if let Some(results) = self.fetch_reqs.get(&rid) {
-                if let session::Protocol::Fetch {
+            if results.send(result).is_err() {
-                    results: Some(results),
+                self.fetch_reqs.remove(&rid);
-                } = protocol
+            }
-                {
+        }
-                    results.send(result).unwrap();
+        if let Some(session) = self.sessions.get_mut(&remote) {
+            if let session::State::Connected { protocol, .. } = &mut session.state {
+                if *protocol == session::Protocol::Fetch {
+                    *protocol = session::Protocol::Gossip;
                 } else {
-                    // Fetch initiated by remote, we don't need to report back.
+                    panic!(
+                        "Unexpected session state for {}: expected 'fetch' protocol, got 'gossip'",
+                        session.id
+                    );
                 }
             }
         }
@@ -924,7 +934,7 @@ where
                 }
             }
             (session::State::Connected { protocol, .. }, Message::Fetch { repo }) => {
-                *protocol = Protocol::Fetch { results: None };
+                *protocol = Protocol::Fetch;
                 // Instruct the transport to handover the socket to the worker.
                 self.reactor
                     .fetch(*remote, repo, Namespaces::default(), false);

radicle-node/src/service/session.rs

@@ -1,7 +1,6 @@
-use crate::service::chan;
 use crate::service::message;
 use crate::service::message::Message;
-use crate::service::{storage, FetchResult};
+use crate::service::storage;
 use crate::service::{Id, LocalTime, NodeId, Reactor, Rng};
 use crate::Link;
 
@@ -17,7 +16,7 @@ pub enum PingState {
 }
 
 /// Session protocol.
-#[derive(Debug, Default, Clone)]
+#[derive(Debug, Default, Copy, PartialEq, Eq, Clone)]
 pub enum Protocol {
     /// The default message-based gossip protocol.
     #[default]
@@ -25,12 +24,7 @@ pub enum Protocol {
     /// Git smart protocol. Used for fetching repository data.
     /// This protocol is used after a connection upgrade via the
     /// [`Message::Fetch`] message.
-    Fetch {
+    Fetch,
-        /// Channel to send fetch results on. Set to `Some` when the fetch
-        /// is locally initiated. Otherwise, no results need to be communicated
-        /// back.
-        results: Option<chan::Sender<FetchResult>>,
-    },
 }
 
 #[derive(Debug, Clone)]
@@ -145,12 +139,10 @@ impl Session {
         self.attempts += 1;
     }
 
-    pub fn fetch(&mut self, repo: Id, results: chan::Sender<FetchResult>) -> Option<Message> {
+    pub fn fetch(&mut self, repo: Id) -> Option<Message> {
         if let State::Connected { protocol, .. } = &mut self.state {
             if let Protocol::Gossip = protocol {
-                *protocol = Protocol::Fetch {
+                *protocol = Protocol::Fetch;
-                    results: Some(results),
-                };
                 return Some(Message::Fetch { repo });
             } else {
                 log::error!(

radicle-node/src/tests/e2e.rs

@@ -18,7 +18,7 @@ use radicle::Storage;
 use radicle::{assert_matches, rad};
 
 use crate::node::NodeId;
-use crate::service::{FetchLookup, FetchResult};
+use crate::service::FetchLookup;
 use crate::storage::git::transport;
 use crate::test::logger;
 use crate::{client, client::handle::Handle, client::Runtime, service};
@@ -343,16 +343,17 @@ fn test_replication() {
     };
     assert_eq!(seeds, nonempty::NonEmpty::new(bob.id));
 
-    let (from, updated) = match results.recv_timeout(Duration::from_secs(6)).unwrap() {
+    let result = results.recv_timeout(Duration::from_secs(6)).unwrap();
-        FetchResult::Fetched { from, updated } => (from, updated),
+    let updated = match result.result {
-        FetchResult::Error { from, error } => {
+        Ok(updated) => updated,
-            panic!("Fetch failed from {from}: {error}");
+        Err(err) => {
+            panic!("Fetch failed from {}: {err}", result.remote);
         }
     };
-    assert_eq!(from, bob.id);
+    assert_eq!(result.remote, bob.id);
     assert_eq!(updated, vec![]);
 
-    log::debug!(target: "test", "Fetch complete with {}", from);
+    log::debug!(target: "test", "Fetch complete with {}", result.remote);
 
     let inventory = alice.handle.inventory().unwrap();
     let alice_refs = alice

radicle-node/src/worker.rs

@@ -57,15 +57,10 @@ impl<G: Signer + EcSign + 'static> Worker<G> {
         } = task;
 
         let (session, result) = self._process(&fetch, drain, session);
-        let result = match result {
+        let result = FetchResult {
-            Ok(updated) => FetchResult::Fetched {
+            rid: fetch.repo,
-                from: fetch.remote,
+            remote: fetch.remote,
-                updated,
+            result,
-            },
-            Err(error) => FetchResult::Error {
-                from: fetch.remote,
-                error,
-            },
         };
         log::debug!(target: "worker", "Sending response back to service..");