node: Always check refs status before fetching

We were only checking it on dequeue.

Also simplifies some of the logic by centralizing the check for fetching
our own refs.

radicle-node/src/service.rs

@@ -830,17 +830,32 @@ where
         }
     }
 
-    /// Initiate an outgoing fetch for some repository, based on
+    /// Initiate an outgoing fetch for some repository, based on another node's announcement.
-    /// another node's announcement.
+    /// Returns `true` if the fetch was initiated and `false` if it was skipped.
     fn fetch_refs_at(
         &mut self,
         rid: RepoId,
         from: NodeId,
         refs: NonEmpty<RefsAt>,
+        scope: Scope,
         timeout: time::Duration,
         channel: Option<chan::Sender<FetchResult>>,
-    ) {
+    ) -> bool {
-        self._fetch(rid, from, refs.into(), timeout, channel)
+        match self.refs_status_of(rid, refs, &scope) {
+            Ok(status) => {
+                if status.want.is_empty() {
+                    debug!(target: "service", "Skipping  fetch for {rid}, all refs are already in storage");
+                } else {
+                    self._fetch(rid, from, status.want, timeout, channel);
+                    return true;
+                }
+            }
+            Err(e) => {
+                error!(target: "service", "Error getting the refs status of {rid}: {e}");
+            }
+        }
+        // We didn't try to fetch anything.
+        false
     }
 
     /// Initiate an outgoing fetch for some repository.
@@ -942,6 +957,7 @@ where
             return Err(TryFetchError::SessionCapacityReached);
         }
 
+        let remotes = refs_at.len();
         let fetching = fetching.or_insert(FetchState {
             from,
             refs_at: refs_at.clone(),
@@ -952,7 +968,10 @@ where
         self.outbox
             .fetch(session, rid, namespaces, refs_at, timeout);
 
-        debug!(target: "service", "Fetch initiated for {rid} with {}..", session.id);
+        debug!(
+            target: "service",
+            "Fetch initiated for {rid} with {} ({remotes} remote(s))..", session.id
+        );
 
         Ok(fetching)
     }
@@ -1077,19 +1096,12 @@ where
                 .seed_policy(&rid)
                 .expect("Service::dequeue_fetch: error accessing repo seeding configuration");
 
-            match self.refs_status_of(rid, refs_at, &repo_entry.scope) {
+            if let Some(refs) = NonEmpty::from_vec(refs_at) {
-                Ok(status) => {
+                if self.fetch_refs_at(rid, from, refs, repo_entry.scope, FETCH_TIMEOUT, channel) {
-                    if let Some(refs) = NonEmpty::from_vec(status.fresh) {
+                    break;
-                        self.fetch_refs_at(rid, from, refs, FETCH_TIMEOUT, channel);
-                        return;
-                    } else {
-                        trace!(target: "service", "Skipping dequeued fetch for {rid}, all refs are already in local storage");
-                    }
-                }
-                Err(e) => {
-                    error!(target: "service", "Error getting the refs status of {rid}: {e}");
-                    return;
                 }
+            } else {
+                self.fetch(rid, from, FETCH_TIMEOUT, channel);
             }
         }
     }
@@ -1475,15 +1487,15 @@ where
                 };
 
                 // Finally, if there's anything to fetch, we fetch it from the remote.
-                if let Some(refs) = NonEmpty::from_vec(
+                if let Some(refs) = NonEmpty::from_vec(message.refs.to_vec()) {
-                    message
+                    self.fetch_refs_at(
-                        .refs
+                        message.rid,
-                        .iter()
+                        remote.id,
-                        .filter(|r| r.remote != self.node_id()) // Don't fetch our own refs.
+                        refs,
-                        .cloned()
+                        repo_entry.scope,
-                        .collect(),
+                        FETCH_TIMEOUT,
-                ) {
+                        None,
-                    self.fetch_refs_at(message.rid, remote.id, refs, FETCH_TIMEOUT, None);
+                    );
                 } else {
                     debug!(target: "service", "Skipping fetch, no remote refs in announcement for {}", message.rid);
                 }
@@ -1685,31 +1697,30 @@ where
     fn refs_status_of(
         &self,
         rid: RepoId,
-        refs: Vec<RefsAt>,
+        refs: NonEmpty<RefsAt>,
         scope: &policy::Scope,
     ) -> Result<RefsStatus, Error> {
         let mut refs = RefsStatus::new(rid, refs, self.db.refs())?;
-        // First, check the freshness.
+        // Check that there's something we want.
-        if refs.fresh.is_empty() {
+        if refs.want.is_empty() {
             return Ok(refs);
         }
-        // Second, check the scope.
+        // Check scope.
-        match scope {
+        let mut refs = match scope {
-            policy::Scope::All => Ok(refs),
+            policy::Scope::All => refs,
-            policy::Scope::Followed => {
+            policy::Scope::Followed => match self.policies.namespaces_for(&self.storage, &rid) {
-                match self.policies.namespaces_for(&self.storage, &rid) {
+                Ok(Namespaces::All) => refs,
-                    Ok(Namespaces::All) => Ok(refs),
+                Ok(Namespaces::Followed(followed)) => {
-                    Ok(Namespaces::Followed(mut followed)) => {
+                    refs.want.retain(|r| followed.contains(&r.remote));
-                        // Get the set of followed nodes except self.
+                    refs
-                        followed.remove(self.nid());
-                        refs.fresh.retain(|r| followed.contains(&r.remote));
-
-                        Ok(refs)
-                    }
-                    Err(e) => Err(e.into()),
                 }
-            }
+                Err(e) => return Err(e.into()),
-        }
+            },
+        };
+        // Remove our own remote, we don't want to fetch that.
+        refs.want.retain(|r| r.remote != self.node_id());
+
+        Ok(refs)
     }
 
     /// Set of initial messages to send to a peer.

radicle-node/src/service/message.rs

@@ -1,5 +1,6 @@
 use std::{fmt, io, mem};
 
+use nonempty::NonEmpty;
 use radicle::git;
 use radicle::storage::refs::RefsAt;
 
@@ -167,14 +168,14 @@ pub struct RefsAnnouncement {
 #[derive(Default)]
 pub struct RefsStatus {
     /// The `rad/sigrefs` was missing or it's ahead of the local
-    /// `rad/sigrefs`.
+    /// `rad/sigrefs`. We want it.
-    pub fresh: Vec<RefsAt>,
+    pub want: Vec<RefsAt>,
-    /// The `rad/sigrefs` has been seen before.
+    /// The `rad/sigrefs` has been seen before. We already have it.
-    pub stale: Vec<RefsAt>,
+    pub have: Vec<RefsAt>,
 }
 
 impl RefsStatus {
-    /// Get the set of `fresh` and `stale` `RefsAt`'s for the given
+    /// Get the set of `want` and `have` `RefsAt`'s for the given
     /// announcement.
     ///
     /// Nb. We use the refs database as a cache for quick lookups. This does *not* check
@@ -183,7 +184,7 @@ impl RefsStatus {
     /// and old refs announcements will be discarded due to their lower timestamps.
     pub fn new<D: node::refs::Store>(
         rid: RepoId,
-        refs: Vec<RefsAt>,
+        refs: NonEmpty<RefsAt>,
         db: &D,
     ) -> Result<RefsStatus, storage::Error> {
         let mut status = RefsStatus::default();
@@ -202,13 +203,13 @@ impl RefsStatus {
         match db.get(repo, &theirs.remote, &storage::refs::SIGREFS_BRANCH) {
             Ok(Some((ours, _))) => {
                 if theirs.at != ours {
-                    self.fresh.push(theirs);
+                    self.want.push(theirs);
                 } else {
-                    self.stale.push(theirs);
+                    self.have.push(theirs);
                 }
             }
             Ok(None) => {
-                self.fresh.push(theirs);
+                self.want.push(theirs);
             }
             Err(e) => {
                 log::warn!(