From 7e13e0759fb66fd91d71be3b88c9c5688407ec9c Mon Sep 17 00:00:00 2001 From: cloudhead Date: Sun, 31 Mar 2024 23:35:14 +0200 Subject: [PATCH] node: Simplify handling of fetch completion * Don't branch based on whether it was a user-requested fetch or not. * Only announce inventory on clones of public repos. --- radicle-cli/tests/commands.rs | 2 + radicle-node/src/service.rs | 126 +++++++++++++++----------------- radicle-node/src/test/handle.rs | 4 +- radicle-node/src/tests.rs | 11 ++- radicle-node/src/tests/e2e.rs | 3 +- radicle/src/identity/doc.rs | 2 +- radicle/src/node.rs | 10 +-- radicle/src/storage.rs | 5 ++ 8 files changed, 79 insertions(+), 84 deletions(-) diff --git a/radicle-cli/tests/commands.rs b/radicle-cli/tests/commands.rs index 46fa80bd..fe3fec87 100644 --- a/radicle-cli/tests/commands.rs +++ b/radicle-cli/tests/commands.rs @@ -389,6 +389,8 @@ fn rad_id_multi_delegate() { eve.fork(acme, working.join("eve")).unwrap(); eve.has_remote_of(&acme, &bob.id); alice.has_remote_of(&acme, &eve.id); + alice.is_synced_with(&acme, &eve.id); + alice.is_synced_with(&acme, &bob.id); // TODO: Have formula with two connected nodes and a tracked project. diff --git a/radicle-node/src/service.rs b/radicle-node/src/service.rs index 23d532a7..8e12f9e4 100644 --- a/radicle-node/src/service.rs +++ b/radicle-node/src/service.rs @@ -963,7 +963,40 @@ where remote: NodeId, result: Result, ) { - let result = match result { + let Some(fetching) = self.fetching.remove(&rid) else { + error!(target: "service", "Received unexpected fetch result for {rid}, from {remote}"); + return; + }; + debug_assert_eq!(fetching.from, remote); + + if let Some(s) = self.sessions.get_mut(&remote) { + // Mark this RID as fetched for this session. + s.fetched(rid); + } + + // Notify all fetch subscribers of the fetch result. This is used when the user requests + // a fetch via the CLI, for example. + for sub in &fetching.subscribers { + debug!(target: "service", "Found existing fetch request from {remote}, sending result.."); + + let result = match &result { + Ok(success) => FetchResult::Success { + updated: success.updated.clone(), + namespaces: success.namespaces.clone(), + clone: success.clone, + }, + Err(e) => FetchResult::Failed { + reason: e.to_string(), + }, + }; + if sub.send(result).is_err() { + error!(target: "service", "Error sending fetch result for {rid} from {remote}.."); + } else { + debug!(target: "service", "Sent fetch result for {rid} from {remote}.."); + } + } + + match result { Ok(fetch::FetchResult { updated, namespaces, @@ -985,81 +1018,35 @@ where updated: updated.clone(), }); - FetchResult::Success { - updated, - namespaces, - clone, - doc, + // Announce our new inventory if this fetch was a full clone. + // Only update and announce inventory for public repositories. + if clone && doc.visibility.is_public() { + debug!(target: "service", "Updating and announcing inventory for cloned repository {rid}.."); + + self.storage.insert(rid); + self.sync_and_announce_inventory(); + } + + // It's possible for a fetch to succeed but nothing was updated. + if updated.is_empty() { + debug!(target: "service", "Nothing to announce, no refs were updated.."); + } else { + // Finally, announce the refs. This is useful for nodes to know what we've synced, + // beyond just knowing that we have added an item to our inventory. + if let Err(e) = self.announce_refs(rid, doc.into(), namespaces) { + error!(target: "service", "Failed to announce new refs: {e}"); + } } } Err(err) => { - let reason = err.to_string(); - error!(target: "service", "Fetch failed for {rid} from {remote}: {reason}"); + error!(target: "service", "Fetch failed for {rid} from {remote}: {err}"); // For now, we only disconnect the remote in case of timeout. In the future, // there may be other reasons to disconnect. if err.is_timeout() { self.outbox.disconnect(remote, DisconnectReason::Fetch(err)); } - FetchResult::Failed { reason } } - }; - - let Some(fetching) = self.fetching.remove(&rid) else { - warn!(target: "service", "Received unexpected fetch result for {rid}, from {remote}"); - return; - }; - debug_assert_eq!(fetching.from, remote); - - if let Some(s) = self.sessions.get_mut(&remote) { - // Mark this RID as fetched for this session. - s.fetched(rid); - } - - for sub in &fetching.subscribers { - debug!(target: "service", "Found existing fetch request from {remote}, sending result.."); - - if sub.send(result.clone()).is_err() { - error!(target: "service", "Error sending fetch result for {rid} from {remote}.."); - } else { - debug!(target: "service", "Sent fetch result for {rid} from {remote}.."); - } - } - - if fetching.subscribers.is_empty() { - trace!(target: "service", "No fetch requests found for {rid}.."); - - // TODO: Combine this with the below. - // We only announce refs here when the fetch wasn't user-requested. This is - // because the user might want to announce his fork, once he has created one, - // or may choose to not announce anything. - match &result { - FetchResult::Success { - updated, - namespaces, - doc, - .. - } if !updated.is_empty() => { - if let Err(e) = - self.announce_refs(rid, doc.clone().into(), namespaces.iter().cloned()) - { - error!(target: "service", "Failed to announce new refs: {e}"); - } - } - _ => debug!(target: "service", "Nothing to announce, no refs were updated.."), - } - } - - // Announce our new inventory if this fetch was a clone. - if let FetchResult::Success { - doc, clone: true, .. - } = result - { - // Only add inventory for public repositories. - if doc.visibility.is_public() { - self.storage.insert(rid); - } - self.sync_and_announce(); } // We can now try to dequeue another fetch. @@ -1898,8 +1885,9 @@ where let peers = self.sessions.connected().map(|(_, p)| p); let (ann, refs) = self.refs_announcement_for(rid, remotes)?; - // Update our local sync status. This is useful for determining if refs were updated while - // the node was stopped. + // Update our sync status for our own refs. This is useful for determining if refs were + // updated while the node was stopped. + // TODO: Move to `announce_own_refs`. if let Some(refs) = refs.iter().find(|r| r.remote == ann.node) { info!(target: "service", "Announcing local refs for {rid} to peers ({})..", refs.at); @@ -1923,7 +1911,7 @@ where Ok(refs) } - fn sync_and_announce(&mut self) { + fn sync_and_announce_inventory(&mut self) { match self.sync_inventory() { Ok(synced) => { // Only announce if our inventory changed. diff --git a/radicle-node/src/test/handle.rs b/radicle-node/src/test/handle.rs index d71736f0..231a7e43 100644 --- a/radicle-node/src/test/handle.rs +++ b/radicle-node/src/test/handle.rs @@ -5,9 +5,8 @@ use std::time; use radicle::git; use radicle::storage::refs::RefsAt; -use radicle::test::arbitrary; -use crate::identity::{DocAt, RepoId}; +use crate::identity::RepoId; use crate::node::{Alias, Config, ConnectOptions, ConnectResult, Event, FetchResult, Seeds}; use crate::runtime::HandleError; use crate::service::policy; @@ -63,7 +62,6 @@ impl radicle::node::Handle for Handle { updated: vec![], namespaces: HashSet::new(), clone: false, - doc: arbitrary::gen::(1), }) } diff --git a/radicle-node/src/tests.rs b/radicle-node/src/tests.rs index 32a5dcfb..90a11bde 100644 --- a/radicle-node/src/tests.rs +++ b/radicle-node/src/tests.rs @@ -1652,21 +1652,24 @@ fn test_init_and_seed() { sim.run_while([&mut alice, &mut bob, &mut eve], |s| !s.is_settled()); + log::debug!(target: "test", "Simulation is over"); + // TODO: Refs should be compared between the two peers. - assert!(eve.storage().get(proj_id).unwrap().is_some()); - assert!(bob.storage().get(proj_id).unwrap().is_some()); - + log::debug!(target: "test", "Waiting for {} to fetch {} from {}..", bob.id, proj_id,eve.id); bob_events .iter() .find(|e| { matches!( e, service::Event::RefsFetched { remote, .. } - if *remote == eve.node_id(), + if *remote == eve.node_id() ) }) .expect("Bob fetched from Eve"); + + assert!(eve.storage().get(proj_id).unwrap().is_some()); + assert!(bob.storage().get(proj_id).unwrap().is_some()); } #[test] diff --git a/radicle-node/src/tests/e2e.rs b/radicle-node/src/tests/e2e.rs index 4375b83d..eeaec58c 100644 --- a/radicle-node/src/tests/e2e.rs +++ b/radicle-node/src/tests/e2e.rs @@ -938,7 +938,8 @@ fn test_non_fastforward_sigrefs() { assert_matches!( alice.handle.fetch(rid, eve.id, DEFAULT_TIMEOUT).unwrap(), - FetchResult::Success { updated, .. } if updated.is_empty() + FetchResult::Success { updated, .. } + if updated.iter().all(|u| u.is_skipped()) ); } diff --git a/radicle/src/identity/doc.rs b/radicle/src/identity/doc.rs index c5ba8319..e1c1670e 100644 --- a/radicle/src/identity/doc.rs +++ b/radicle/src/identity/doc.rs @@ -130,7 +130,7 @@ impl Deref for Payload { } /// A verified identity document at a specific commit. -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Eq)] pub struct DocAt { /// The commit at which this document exists. pub commit: Oid, diff --git a/radicle/src/node.rs b/radicle/src/node.rs index a05ac9c8..d2bfd793 100644 --- a/radicle/src/node.rs +++ b/radicle/src/node.rs @@ -29,7 +29,7 @@ use serde_json as json; use crate::crypto::PublicKey; use crate::git; -use crate::identity::{DocAt, RepoId}; +use crate::identity::RepoId; use crate::profile; use crate::storage::refs::RefsAt; use crate::storage::RefUpdate; @@ -686,7 +686,6 @@ pub enum FetchResult { updated: Vec, namespaces: HashSet, clone: bool, - doc: DocAt, }, // TODO: Create enum for reason. Failed { @@ -719,14 +718,13 @@ impl FetchResult { } } -impl From, HashSet, bool, DocAt), S>> for FetchResult { - fn from(value: Result<(Vec, HashSet, bool, DocAt), S>) -> Self { +impl From, HashSet, bool), S>> for FetchResult { + fn from(value: Result<(Vec, HashSet, bool), S>) -> Self { match value { - Ok((updated, namespaces, clone, doc)) => Self::Success { + Ok((updated, namespaces, clone)) => Self::Success { updated, namespaces, clone, - doc, }, Err(err) => Self::Failed { reason: err.to_string(), diff --git a/radicle/src/storage.rs b/radicle/src/storage.rs index 6d034abe..d3e87436 100644 --- a/radicle/src/storage.rs +++ b/radicle/src/storage.rs @@ -239,6 +239,11 @@ impl RefUpdate { pub fn is_created(&self) -> bool { matches!(self, RefUpdate::Created { .. }) } + + /// Is it a skip. + pub fn is_skipped(&self) -> bool { + matches!(self, RefUpdate::Skipped { .. }) + } } impl fmt::Display for RefUpdate {