node: duplicate fetching logic respects RefsAt — heartwood

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.

 struct FetchState {
     /// Node we're fetching from.
     from: NodeId,
     /// What refs we're fetch from
     refs_at: Vec<RefsAt>,
     /// Channels waiting for fetch results.
     subscribers: Vec<chan::Sender<FetchResult>>,
 }

                 }
             }
             Err(TryFetchError::AlreadyFetching(fetching)) => {
                 // If we're already fetching from the requested peer, there's nothing to do, we
                 // simply add the supplied channel to the list of subscribers so that it is notified
                 // on completion. Otherwise, we queue a fetch with the requested peer.
                 if fetching.from == from {
                 // If we're already fetching the same refs from the
                 // requested peer, there's nothing to do, we simply
                 // add the supplied channel to the list of subscribers
                 // so that it is notified on completion. Otherwise, we
                 // queue a fetch with the requested peer.
                 if fetching.from == from && fetching.refs_at == refs_at {
                     debug!(target: "service", "Ignoring redundant fetch of {rid} from {from}");
                     if let Some(c) = channel {

         let fetching = fetching.or_insert(FetchState {
             from,
             refs_at: refs_at.clone(),
             subscribers: vec![],
         });
         let namespaces = self.policies.namespaces_for(&self.storage, &rid)?;

 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,

 };
 use radicle::test::fixtures;
 use radicle::{assert_matches, rad};
 use radicle::{git, issue};
 use crate::node::config::Limits;
 use crate::node::{Config, ConnectOptions};

     // 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",

         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);