node: Ensure dequeued fetches are correct

Dequeued fetches were always fetching all refs, instead of fetching the
specific refs that were initially requested. We fix that by storing the
refs to fetch in the fetch queue.

radicle-node/src/service.rs

@@ -264,6 +264,8 @@ struct QueuedFetch {
     rid: RepoId,
     /// Peer being fetched from.
     from: NodeId,
+    /// Refs being fetched.
+    refs_at: Vec<RefsAt>,
     /// Result channel.
     channel: Option<chan::Sender<FetchResult>>,
 }
@@ -791,8 +793,9 @@ where
         from: NodeId,
         refs: NonEmpty<RefsAt>,
         timeout: time::Duration,
+        channel: Option<chan::Sender<FetchResult>>,
     ) {
-        self._fetch(rid, from, refs.into(), timeout, None)
+        self._fetch(rid, from, refs.into(), timeout, channel)
     }
 
     /// Initiate an outgoing fetch for some repository.
@@ -814,7 +817,7 @@ where
         timeout: time::Duration,
         channel: Option<chan::Sender<FetchResult>>,
     ) {
-        match self.try_fetch(rid, &from, refs_at, timeout) {
+        match self.try_fetch(rid, &from, refs_at.clone(), timeout) {
             Ok(fetching) => {
                 if let Some(c) = channel {
                     fetching.subscribe(c);
@@ -832,12 +835,22 @@ where
                     }
                 } else {
                     debug!(target: "service", "Queueing fetch for {rid} with {from}..");
-                    self.queue.push_back(QueuedFetch { rid, from, channel });
+                    self.queue.push_back(QueuedFetch {
+                        rid,
+                        refs_at,
+                        from,
+                        channel,
+                    });
                 }
             }
             Err(TryFetchError::SessionCapacityReached) => {
                 debug!(target: "service", "Fetch capacity reached for {from}, queueing {rid}..");
-                self.queue.push_back(QueuedFetch { rid, from, channel });
+                self.queue.push_back(QueuedFetch {
+                    rid,
+                    refs_at,
+                    from,
+                    channel,
+                });
             }
             Err(e) => {
                 if let Some(c) = channel {
@@ -850,6 +863,7 @@ where
         }
     }
 
+    // TODO: Buffer/throttle fetches.
     fn try_fetch(
         &mut self,
         rid: RepoId,
@@ -863,6 +877,8 @@ where
         };
         let fetching = self.fetching.entry(rid);
 
+        trace!(target: "service", "Trying to fetch {refs_at:?} for {rid}..");
+
         if let Entry::Occupied(fetching) = fetching {
             // We're already fetching this repo from some peer.
             return Err(TryFetchError::AlreadyFetching(fetching.into_mut()));
@@ -991,10 +1007,40 @@ where
     /// 1. The RID was already being fetched.
     /// 2. The session was already at fetch capacity.
     pub fn dequeue_fetch(&mut self) {
-        if let Some(QueuedFetch { rid, from, channel }) = self.queue.pop_front() {
+        while let Some(QueuedFetch {
+            rid,
+            from,
+            refs_at,
+            channel,
+        }) = self.queue.pop_front()
+        {
             debug!(target: "service", "Dequeued fetch for {rid} from session {from}..");
 
-            self.fetch(rid, from, FETCH_TIMEOUT, channel);
+            // If no refs are specified, always do a full fetch.
+            if refs_at.is_empty() {
+                self.fetch(rid, from, FETCH_TIMEOUT, channel);
+                return;
+            }
+
+            let repo_entry = self
+                .policies
+                .seed_policy(&rid)
+                .expect("Service::dequeue_fetch: error accessing repo seeding configuration");
+
+            match self.refs_status_of(rid, refs_at, &repo_entry.scope) {
+                Ok(status) => {
+                    if let Some(refs) = NonEmpty::from_vec(status.fresh) {
+                        self.fetch_refs_at(rid, from, refs, FETCH_TIMEOUT, channel);
+                        return;
+                    } else {
+                        debug!(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;
+                }
+            }
         }
     }
 
@@ -1330,13 +1376,15 @@ where
                     }
                 }
 
-                // TODO: Buffer/throttle fetches.
                 let repo_entry = self.policies.seed_policy(&message.rid).expect(
                     "Service::handle_announcement: error accessing repo seeding configuration",
                 );
-
                 if repo_entry.policy == Policy::Allow {
-                    let (fresh, stale) = match self.refs_status_of(message, &repo_entry.scope) {
+                    let (fresh, stale) = match self.refs_status_of(
+                        message.rid,
+                        message.refs.clone().into(),
+                        &repo_entry.scope,
+                    ) {
                         Ok(RefsStatus { fresh, stale }) => (fresh, stale),
                         Err(e) => {
                             error!(target: "service", "Failed to check refs status: {e}");
@@ -1392,7 +1440,9 @@ where
                         // Finally, if there's anything to fetch, we fetch it from the
                         // remote.
                         if let Some(fresh) = NonEmpty::from_vec(fresh) {
-                            self.fetch_refs_at(message.rid, remote.id, fresh, FETCH_TIMEOUT);
+                            self.fetch_refs_at(message.rid, remote.id, fresh, FETCH_TIMEOUT, None);
+                        } else {
+                            debug!(target: "service", "Skipping fetch, all refs of {} are already in local storage", message.rid);
                         }
                     } else {
                         trace!(
@@ -1479,14 +1529,14 @@ where
     /// A convenient method to check if we should fetch from a `RefsAnnouncement` with `scope`.
     fn refs_status_of(
         &self,
-        message: &RefsAnnouncement,
+        rid: RepoId,
+        refs: Vec<RefsAt>,
         scope: &policy::Scope,
     ) -> Result<RefsStatus, Error> {
-        let mut refs = message.refs_status(&self.storage)?;
+        let mut refs = RefsStatus::new(rid, refs, &self.storage)?;
 
         // First, check the freshness.
         if refs.fresh.is_empty() {
-            debug!(target: "service", "All refs of {} are already in local storage", &message.rid);
             return Ok(refs);
         }
 
@@ -1494,7 +1544,7 @@ where
         match scope {
             policy::Scope::All => Ok(refs),
             policy::Scope::Followed => {
-                match self.policies.namespaces_for(&self.storage, &message.rid) {
+                match self.policies.namespaces_for(&self.storage, &rid) {
                     Ok(Namespaces::All) => Ok(refs),
                     Ok(Namespaces::Followed(mut followed)) => {
                         // Get the set of followed nodes except self.

radicle-node/src/service/message.rs

@@ -176,6 +176,34 @@ pub struct RefsStatus {
 }
 
 impl RefsStatus {
+    /// Get the set of `fresh` and `stale` `RefsAt`'s for the given
+    /// announcement.
+    pub fn new<S: ReadStorage>(
+        rid: RepoId,
+        refs: Vec<RefsAt>,
+        storage: S,
+    ) -> Result<RefsStatus, storage::Error> {
+        let repo = match storage.repository(rid) {
+            // If the repo doesn't exist, we consider this
+            // announcement "fresh", since we obviously don't
+            // have the refs.
+            Err(e) if e.is_not_found() => {
+                return Ok(RefsStatus {
+                    fresh: refs.clone(),
+                    stale: Vec::new(),
+                })
+            }
+            Err(e) => return Err(e),
+            Ok(r) => r,
+        };
+
+        let mut status = RefsStatus::default();
+        for theirs in refs.iter() {
+            status.insert(*theirs, &repo)?;
+        }
+        Ok(status)
+    }
+
     fn insert<S: ReadRepository>(
         &mut self,
         theirs: RefsAt,
@@ -216,32 +244,6 @@ impl RefsStatus {
     }
 }
 
-impl RefsAnnouncement {
-    /// Get the set of `fresh` and `stale` `RefsAt`'s for the given
-    /// announcement.
-    pub fn refs_status<S: ReadStorage>(&self, storage: S) -> Result<RefsStatus, storage::Error> {
-        let repo = match storage.repository(self.rid) {
-            // If the repo doesn't exist, we consider this
-            // announcement "fresh", since we obviously don't
-            // have the refs.
-            Err(e) if e.is_not_found() => {
-                return Ok(RefsStatus {
-                    fresh: self.refs.clone().into(),
-                    stale: Vec::new(),
-                })
-            }
-            Err(e) => return Err(e),
-            Ok(r) => r,
-        };
-
-        let mut status = RefsStatus::default();
-        for theirs in self.refs.iter() {
-            status.insert(*theirs, &repo)?;
-        }
-        Ok(status)
-    }
-}
-
 /// Node announcing its inventory to the network.
 /// This should be the whole inventory every time.
 #[derive(Debug, Clone, PartialEq, Eq)]

radicle-node/src/test/peer.rs

@@ -11,7 +11,7 @@ use radicle::node::address::Store as _;
 use radicle::node::Database;
 use radicle::node::{address, Alias, ConnectOptions};
 use radicle::rad;
-use radicle::storage::refs::RefsAt;
+use radicle::storage::refs::{RefsAt, SignedRefsAt};
 use radicle::storage::{ReadRepository, RemoteRepository};
 use radicle::Storage;
 
@@ -319,14 +319,22 @@ where
             }
         }
 
-        let ann = AnnouncementMessage::from(RefsAnnouncement {
+        self.announcement(RefsAnnouncement {
             rid,
             refs,
             timestamp: self.timestamp(),
-        });
+        })
-        let msg = ann.signed(self.signer());
+    }
 
-        msg.into()
+    pub fn announcement(&self, ann: impl Into<AnnouncementMessage>) -> Message {
+        ann.into().signed(self.signer()).into()
+    }
+
+    pub fn signed_refs_at(&self, refs: Refs, at: radicle::git::Oid) -> SignedRefsAt {
+        SignedRefsAt {
+            sigrefs: refs.signed(self.signer()).unwrap(),
+            at,
+        }
     }
 
     pub fn connect_from(&mut self, peer: &Self) {

radicle-node/src/tests.rs

@@ -13,6 +13,7 @@ use radicle::node::address::Store;
 use radicle::node::routing::Store as _;
 use radicle::node::{ConnectOptions, DEFAULT_TIMEOUT};
 use radicle::storage::refs::RefsAt;
+use radicle::storage::RefUpdate;
 
 use crate::collections::{RandomMap, RandomSet};
 use crate::crypto::test::signer::MockSigner;
@@ -38,7 +39,6 @@ use crate::test::peer;
 use crate::test::peer::Peer;
 use crate::test::simulator;
 use crate::test::simulator::{Peer as _, Simulation};
-use crate::test::storage as mock_storage;
 use crate::test::storage::MockStorage;
 use crate::wire::Decode;
 use crate::wire::Encode;
@@ -763,37 +763,24 @@ fn test_refs_announcement_followed() {
 
     // Create MockStorage for Alice and Bob. Both will have repo with `rid`.
     let storage_alice = arbitrary::nonempty_storage(1);
-    let rid = *storage_alice.inventory.keys().next().unwrap();
+    let rid = *storage_alice.repos.keys().next().unwrap();
     let storage_bob = storage_alice.clone();
     let mut alice = Peer::with_storage("alice", [7, 7, 7, 7], storage_alice);
     let mut bob = Peer::with_storage("bob", [8, 8, 8, 8], storage_bob);
 
-    let refs = arbitrary::gen::<Refs>(8);
-    let sigref_at = arbitrary::oid();
-    let signed_refs = refs.signed(bob.signer()).unwrap();
     let node_id = alice.id;
-    alice.storage_mut().insert_remote(
+    alice.storage_mut().repo_mut(&rid).remotes.insert(
-        rid,
         node_id,
-        mock_storage::refs::SignedRefsAt {
+        bob.signed_refs_at(arbitrary::gen::<Refs>(8), arbitrary::oid()),
-            at: sigref_at,
-            sigrefs: signed_refs,
-        },
     );
 
     // Generate some refs for Bob under their own node_id.
-    let refs = arbitrary::gen::<Refs>(8);
+    let sigrefs = bob.signed_refs_at(arbitrary::gen::<Refs>(8), arbitrary::oid());
-    let sigref_at = arbitrary::oid();
-    let signed_refs = refs.signed(bob.signer()).unwrap();
     let node_id = bob.id;
-    bob.storage_mut().insert_remote(
+    bob.storage_mut()
-        rid,
+        .repo_mut(&rid)
-        node_id,
+        .remotes
-        mock_storage::refs::SignedRefsAt {
+        .insert(node_id, sigrefs);
-            at: sigref_at,
-            sigrefs: signed_refs,
-        },
-    );
 
     // Alice uses Scope::Followed, and did not track Bob yet.
     alice.connect_to(&bob);
@@ -827,7 +814,7 @@ fn test_refs_announcement_followed() {
 #[test]
 fn test_refs_announcement_no_subscribe() {
     let storage = arbitrary::nonempty_storage(1);
-    let rid = *storage.inventory.keys().next().unwrap();
+    let rid = *storage.repos.keys().next().unwrap();
     let mut alice = Peer::with_storage("alice", [7, 7, 7, 7], storage);
     let bob = Peer::new("bob", [8, 8, 8, 8]);
     let eve = Peer::new("eve", [9, 9, 9, 9]);
@@ -1330,7 +1317,7 @@ fn test_fetch_missing_inventory_on_schedule() {
 #[test]
 fn test_queued_fetch_max_capacity() {
     let storage = arbitrary::nonempty_storage(3);
-    let mut repo_keys = storage.inventory.keys();
+    let mut repo_keys = storage.repos.keys();
     let rid1 = *repo_keys.next().unwrap();
     let rid2 = *repo_keys.next().unwrap();
     let rid3 = *repo_keys.next().unwrap();
@@ -1375,9 +1362,77 @@ fn test_queued_fetch_max_capacity() {
 }
 
 #[test]
-fn test_queued_fetch_same_rid() {
+fn test_queued_fetch_from_ann_same_rid() {
     let storage = arbitrary::nonempty_storage(3);
-    let mut repo_keys = storage.inventory.keys();
+    let mut repo_keys = storage.repos.keys();
+    let rid = *repo_keys.next().unwrap();
+    let mut alice = Peer::with_storage("alice", [7, 7, 7, 7], storage);
+    let bob = Peer::new("bob", [8, 8, 8, 8]);
+    let eve = Peer::new("eve", [9, 9, 9, 9]);
+    let carol = Peer::new("carol", [10, 10, 10, 10]);
+    let oid = arbitrary::oid();
+    let ann = RefsAnnouncement {
+        rid,
+        refs: vec![RefsAt {
+            remote: carol.id(),
+            at: oid,
+        }]
+        .try_into()
+        .unwrap(),
+        timestamp: bob.timestamp(),
+    };
+
+    logger::init(log::Level::Trace);
+
+    alice.seed(&rid, policy::Scope::All).unwrap();
+    alice.connect_to(&bob);
+    alice.connect_to(&eve);
+    alice.connect_to(&carol);
+
+    // Send the first announcement.
+    alice.receive(bob.id, bob.announcement(ann.clone()));
+    // Send the 2nd announcement that will be queued.
+    alice.receive(eve.id, eve.announcement(ann.clone()));
+    // Send the 3rd announcement that will be queued.
+    alice.receive(carol.id, carol.announcement(ann));
+
+    // The first fetch is initiated.
+    assert_matches!(alice.fetches().next(), Some((rid_, nid_, _)) if rid_ == rid && nid_ == bob.id);
+    // We shouldn't send out the 2nd, 3rd fetch while we're doing the 1st fetch.
+    assert_matches!(alice.outbox().next(), None);
+
+    // Have enough time pass that Alice sends a "ping" to Bob.
+    alice.elapse(KEEP_ALIVE_DELTA);
+
+    let refname = carol
+        .id()
+        .to_namespace()
+        .join(git::refname!("refs/sigrefs"));
+
+    // Finish the 1st fetch.
+    alice.storage_mut().repo_mut(&rid).remotes.insert(
+        carol.id(),
+        carol.signed_refs_at(arbitrary::gen::<Refs>(1), oid),
+    );
+    alice.fetched(
+        rid,
+        bob.id,
+        Ok(fetch::FetchResult {
+            updated: vec![RefUpdate::Created {
+                name: refname.clone(),
+                oid,
+            }],
+            namespaces: [carol.id()].into_iter().collect(),
+        }),
+    );
+    // Now the 1st fetch is done, but the 2nd and 3rd fetches are redundant.
+    assert_matches!(alice.fetches().next(), None);
+}
+
+#[test]
+fn test_queued_fetch_from_command_same_rid() {
+    let storage = arbitrary::nonempty_storage(3);
+    let mut repo_keys = storage.repos.keys();
     let rid1 = *repo_keys.next().unwrap();
     let mut alice = Peer::with_storage("alice", [7, 7, 7, 7], storage);
     let bob = Peer::new("bob", [8, 8, 8, 8]);
@@ -1604,9 +1659,9 @@ fn prop_inventory_exchange_dense() {
         let mut routing = RandomMap::with_hasher(rng.clone().into());
 
         for (inv, peer) in &[
-            (alice_inv.inventory, alice.node_id()),
+            (alice_inv.repos, alice.node_id()),
-            (bob_inv.inventory, bob.node_id()),
+            (bob_inv.repos, bob.node_id()),
-            (eve_inv.inventory, eve.node_id()),
+            (eve_inv.repos, eve.node_id()),
         ] {
             for id in inv.keys() {
                 routing

radicle/src/test/arbitrary.rs

@@ -1,4 +1,4 @@
-use std::collections::{BTreeMap, BTreeSet, HashSet};
+use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
 use std::hash::Hash;
 use std::ops::RangeBounds;
 use std::str::FromStr;
@@ -68,7 +68,15 @@ pub fn vec<T: Eq + Arbitrary>(size: usize) -> Vec<T> {
 pub fn nonempty_storage(size: usize) -> MockStorage {
     let mut storage = gen::<MockStorage>(size);
     for _ in 0..size {
-        storage.inventory.insert(gen::<RepoId>(1), gen::<DocAt>(1));
+        let id = gen::<RepoId>(1);
+        storage.repos.insert(
+            id,
+            MockRepository {
+                id,
+                doc: gen::<DocAt>(1),
+                remotes: HashMap::new(),
+            },
+        );
     }
     storage
 }

radicle/src/test/storage.rs

@@ -16,12 +16,11 @@ use super::fixtures;
 #[derive(Clone, Debug)]
 pub struct MockStorage {
     pub path: PathBuf,
-    pub inventory: HashMap<RepoId, DocAt>,
     pub info: git::UserInfo,
 
     /// All refs keyed by RID.
     /// Each value is a map of refs keyed by node Id (public key).
-    pub remotes: HashMap<RepoId, HashMap<NodeId, refs::SignedRefsAt>>,
+    pub repos: HashMap<RepoId, MockRepository>,
 }
 
 impl MockStorage {
@@ -29,19 +28,31 @@ impl MockStorage {
         Self {
             path: PathBuf::default(),
             info: fixtures::user(),
-            inventory: inventory.into_iter().collect(),
+            repos: inventory
-            remotes: HashMap::new(),
+                .into_iter()
+                .map(|(id, doc)| {
+                    (
+                        id,
+                        MockRepository {
+                            id,
+                            doc,
+                            remotes: HashMap::new(),
+                        },
+                    )
+                })
+                .collect(),
         }
     }
 
+    pub fn repo_mut(&mut self, rid: &RepoId) -> &mut MockRepository {
+        self.repos
+            .get_mut(rid)
+            .expect("MockStorage::repo_mut: repository does not exist")
+    }
+
     pub fn empty() -> Self {
         Self::new(Vec::new())
     }
-
-    /// Add a remote `node` with `signed_refs` for the repo `rid`.
-    pub fn insert_remote(&mut self, rid: RepoId, node: NodeId, refs: refs::SignedRefsAt) {
-        self.remotes.entry(rid).or_default().insert(node, refs);
-    }
 }
 
 impl ReadStorage for MockStorage {
@@ -60,23 +71,18 @@ impl ReadStorage for MockStorage {
     }
 
     fn contains(&self, rid: &RepoId) -> Result<bool, RepositoryError> {
-        Ok(self.inventory.contains_key(rid))
+        Ok(self.repos.contains_key(rid))
     }
 
     fn inventory(&self) -> Result<Inventory, Error> {
-        Ok(self.inventory.keys().cloned().collect::<Vec<_>>())
+        Ok(self.repos.keys().cloned().collect::<Vec<_>>())
     }
 
     fn repository(&self, rid: RepoId) -> Result<Self::Repository, Error> {
-        let doc = self
+        self.repos
-            .inventory
             .get(&rid)
-            .ok_or_else(|| Error::Io(io::Error::from(io::ErrorKind::NotFound)))?;
+            .ok_or_else(|| Error::Io(io::Error::from(io::ErrorKind::NotFound)))
-        Ok(MockRepository {
+            .cloned()
-            id: rid,
-            doc: doc.clone(),
-            remotes: self.remotes.get(&rid).cloned().unwrap_or_default(),
-        })
     }
 }
 
@@ -84,12 +90,10 @@ impl WriteStorage for MockStorage {
     type RepositoryMut = MockRepository;
 
     fn repository_mut(&self, rid: RepoId) -> Result<Self::RepositoryMut, Error> {
-        let doc = self.inventory.get(&rid).unwrap();
+        self.repos
-        Ok(MockRepository {
+            .get(&rid)
-            id: rid,
+            .ok_or(Error::Io(io::ErrorKind::NotFound.into()))
-            doc: doc.clone(),
+            .cloned()
-            remotes: self.remotes.get(&rid).cloned().unwrap_or_default(),
-        })
     }
 
     fn create(&self, _rid: RepoId) -> Result<Self::RepositoryMut, Error> {
@@ -103,9 +107,9 @@ impl WriteStorage for MockStorage {
 
 #[derive(Clone, Debug)]
 pub struct MockRepository {
-    id: RepoId,
+    pub id: RepoId,
-    doc: DocAt,
+    pub doc: DocAt,
-    remotes: HashMap<NodeId, refs::SignedRefsAt>,
+    pub remotes: HashMap<NodeId, refs::SignedRefsAt>,
 }
 
 impl MockRepository {