node: duplicate fetching logic respects RefsAt

Two concurrent fetches can occur from the same peer, while also being
for a different set of references. This was seen in the
`test_outdated_sigrefs` test. Alice would have an ongoing fetch from
Eve, and concurrently, Eve will create a new issue -- thus creating a
new `RefsAt` message -- and Alice would attempt to fetch this.

The previous logic would only check that Alice is fetching from Eve
and ignore the new updates -- meaning the test could fail. However, if
the `RefsAt` are checked then the other fetch can be recognised as
new, and queue it.

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

radicle-node/src/service.rs

@@ -249,6 +249,8 @@ enum TryFetchError<'a> {
 struct FetchState {
     /// Node we're fetching from.
     from: NodeId,
+    /// What refs we're fetching.
+    refs_at: Vec<RefsAt>,
     /// Channels waiting for fetch results.
     subscribers: Vec<chan::Sender<FetchResult>>,
 }
@@ -837,10 +839,10 @@ where
                 }
             }
             Err(TryFetchError::AlreadyFetching(fetching)) => {
-                // If we're already fetching from the requested peer, there's nothing to do, we
+                // If we're already fetching the same refs from the requested peer, there's nothing
-                // simply add the supplied channel to the list of subscribers so that it is notified
+                // to do, we simply add the supplied channel to the list of subscribers so that it
-                // on completion. Otherwise, we queue a fetch with the requested peer.
+                // is notified on completion. Otherwise, we queue a fetch with the requested peer.
-                if fetching.from == from {
+                if fetching.from == from && fetching.refs_at == refs_at {
                     debug!(target: "service", "Ignoring redundant fetch of {rid} from {from}");
 
                     if let Some(c) = channel {
@@ -912,6 +914,7 @@ where
 
         let fetching = fetching.or_insert(FetchState {
             from,
+            refs_at: refs_at.clone(),
             subscribers: vec![],
         });
         let namespaces = self.policies.namespaces_for(&self.storage, &rid)?;

radicle-node/src/tests/e2e.rs

@@ -1,7 +1,6 @@
 use std::{collections::HashSet, thread, time};
 
 use radicle::crypto::{test::signer::MockSigner, Signer};
-use radicle::git;
 use radicle::node::{Alias, FetchResult, Handle as _, DEFAULT_TIMEOUT};
 use radicle::storage::{
     ReadRepository, ReadStorage, RefUpdate, RemoteRepository, SignRepository, ValidateRepository,
@@ -9,6 +8,7 @@ use radicle::storage::{
 };
 use radicle::test::fixtures;
 use radicle::{assert_matches, rad};
+use radicle::{git, issue};
 
 use crate::node::config::Limits;
 use crate::node::{Config, ConnectOptions};
@@ -969,7 +969,7 @@ fn test_outdated_sigrefs() {
     // At this stage, Alice and Bob have Eve's fork and Eve does not
     // have Bob's fork
 
-    eve.issue(
+    let issue_id = eve.issue(
         rid,
         "Outdated Sigrefs",
         "Outdated sigrefs are harshing my vibes",
@@ -984,6 +984,11 @@ fn test_outdated_sigrefs() {
         FetchResult::Success { .. }
     );
     let repo = alice.storage.repository(rid).unwrap();
+    let issues = issue::Issues::open(&repo).unwrap();
+    assert!(
+        issues.get(&issue_id).unwrap().is_some(),
+        "Alice did not fetch issue {issue_id}"
+    );
     let eve_remote = repo.remote(&eve.id).unwrap();
     let eves_refs_expected = eve_remote.refs;
     assert_ne!(eves_refs_expected, old_refs);