diff --git a/radicle-node/src/service.rs b/radicle-node/src/service.rs index de2f36ac..e4c77a8d 100644 --- a/radicle-node/src/service.rs +++ b/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, /// Result channel. channel: Option>, } @@ -791,8 +793,9 @@ where from: NodeId, refs: NonEmpty, timeout: time::Duration, + channel: Option>, ) { - 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>, ) { - 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, scope: &policy::Scope, ) -> Result { - 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. diff --git a/radicle-node/src/service/message.rs b/radicle-node/src/service/message.rs index 9cec8a3d..be8e686a 100644 --- a/radicle-node/src/service/message.rs +++ b/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( + rid: RepoId, + refs: Vec, + storage: S, + ) -> Result { + 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( &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(&self, storage: S) -> Result { - 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)] diff --git a/radicle-node/src/test/peer.rs b/radicle-node/src/test/peer.rs index 97863f44..82b02ba6 100644 --- a/radicle-node/src/test/peer.rs +++ b/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) -> 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) { diff --git a/radicle-node/src/tests.rs b/radicle-node/src/tests.rs index e6392333..d26ea3c7 100644 --- a/radicle-node/src/tests.rs +++ b/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::(8); - let sigref_at = arbitrary::oid(); - let signed_refs = refs.signed(bob.signer()).unwrap(); let node_id = alice.id; - alice.storage_mut().insert_remote( - rid, + alice.storage_mut().repo_mut(&rid).remotes.insert( node_id, - mock_storage::refs::SignedRefsAt { - at: sigref_at, - sigrefs: signed_refs, - }, + bob.signed_refs_at(arbitrary::gen::(8), arbitrary::oid()), ); // Generate some refs for Bob under their own node_id. - let refs = arbitrary::gen::(8); - let sigref_at = arbitrary::oid(); - let signed_refs = refs.signed(bob.signer()).unwrap(); + let sigrefs = bob.signed_refs_at(arbitrary::gen::(8), arbitrary::oid()); let node_id = bob.id; - bob.storage_mut().insert_remote( - rid, - node_id, - mock_storage::refs::SignedRefsAt { - at: sigref_at, - sigrefs: signed_refs, - }, - ); + bob.storage_mut() + .repo_mut(&rid) + .remotes + .insert(node_id, sigrefs); // 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::(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()), - (bob_inv.inventory, bob.node_id()), - (eve_inv.inventory, eve.node_id()), + (alice_inv.repos, alice.node_id()), + (bob_inv.repos, bob.node_id()), + (eve_inv.repos, eve.node_id()), ] { for id in inv.keys() { routing diff --git a/radicle/src/test/arbitrary.rs b/radicle/src/test/arbitrary.rs index ac312207..6bad684f 100644 --- a/radicle/src/test/arbitrary.rs +++ b/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(size: usize) -> Vec { pub fn nonempty_storage(size: usize) -> MockStorage { let mut storage = gen::(size); for _ in 0..size { - storage.inventory.insert(gen::(1), gen::(1)); + let id = gen::(1); + storage.repos.insert( + id, + MockRepository { + id, + doc: gen::(1), + remotes: HashMap::new(), + }, + ); } storage } diff --git a/radicle/src/test/storage.rs b/radicle/src/test/storage.rs index efed48ac..b5385c9f 100644 --- a/radicle/src/test/storage.rs +++ b/radicle/src/test/storage.rs @@ -16,12 +16,11 @@ use super::fixtures; #[derive(Clone, Debug)] pub struct MockStorage { pub path: PathBuf, - pub inventory: HashMap, 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>, + pub repos: HashMap, } impl MockStorage { @@ -29,19 +28,31 @@ impl MockStorage { Self { path: PathBuf::default(), info: fixtures::user(), - inventory: inventory.into_iter().collect(), - remotes: HashMap::new(), + repos: inventory + .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 { - Ok(self.inventory.contains_key(rid)) + Ok(self.repos.contains_key(rid)) } fn inventory(&self) -> Result { - Ok(self.inventory.keys().cloned().collect::>()) + Ok(self.repos.keys().cloned().collect::>()) } fn repository(&self, rid: RepoId) -> Result { - let doc = self - .inventory + self.repos .get(&rid) - .ok_or_else(|| Error::Io(io::Error::from(io::ErrorKind::NotFound)))?; - Ok(MockRepository { - id: rid, - doc: doc.clone(), - remotes: self.remotes.get(&rid).cloned().unwrap_or_default(), - }) + .ok_or_else(|| Error::Io(io::Error::from(io::ErrorKind::NotFound))) + .cloned() } } @@ -84,12 +90,10 @@ impl WriteStorage for MockStorage { type RepositoryMut = MockRepository; fn repository_mut(&self, rid: RepoId) -> Result { - let doc = self.inventory.get(&rid).unwrap(); - Ok(MockRepository { - id: rid, - doc: doc.clone(), - remotes: self.remotes.get(&rid).cloned().unwrap_or_default(), - }) + self.repos + .get(&rid) + .ok_or(Error::Io(io::ErrorKind::NotFound.into())) + .cloned() } fn create(&self, _rid: RepoId) -> Result { @@ -103,9 +107,9 @@ impl WriteStorage for MockStorage { #[derive(Clone, Debug)] pub struct MockRepository { - id: RepoId, - doc: DocAt, - remotes: HashMap, + pub id: RepoId, + pub doc: DocAt, + pub remotes: HashMap, } impl MockRepository {