From 269d21deca0f97d9468205074d4ee2ed17148818 Mon Sep 17 00:00:00 2001 From: cloudhead Date: Mon, 13 Nov 2023 12:06:37 +0100 Subject: [PATCH] node: Rewrite `seeds` method to use sync-status We update the `seeds` method to use not only the routing table, but also the sync status information for our owned refs. --- radicle-node/src/service.rs | 63 +++++++++++++++++++++++-------------- radicle/src/node.rs | 42 +++++++++++++++++++++++-- 2 files changed, 79 insertions(+), 26 deletions(-) diff --git a/radicle-node/src/service.rs b/radicle-node/src/service.rs index bfb5f404..6eb3507d 100644 --- a/radicle-node/src/service.rs +++ b/radicle-node/src/service.rs @@ -32,7 +32,7 @@ use crate::crypto::{Signer, Verified}; use crate::identity::{Doc, Id}; use crate::node::routing; use crate::node::routing::InsertResult; -use crate::node::{Address, Alias, Features, FetchResult, HostName, Seed, Seeds}; +use crate::node::{Address, Alias, Features, FetchResult, HostName, Seed, Seeds, SyncStatus}; use crate::prelude::*; use crate::runtime::Emitter; use crate::service::message::{Announcement, AnnouncementMessage, Ping}; @@ -122,6 +122,8 @@ pub enum Error { #[error(transparent)] Routing(#[from] routing::Error), #[error(transparent)] + Addresses(#[from] address::Error), + #[error(transparent)] Tracking(#[from] tracking::Error), #[error(transparent)] Repository(#[from] radicle::storage::RepositoryError), @@ -544,7 +546,7 @@ where resp.send(seeds).ok(); } Err(e) => { - error!(target: "service", "Error reading routing table for {rid}: {e}"); + error!(target: "service", "Error getting seeds for {rid}: {e}"); } }, Command::Fetch(rid, seed, timeout, resp) => { @@ -1445,31 +1447,46 @@ where } fn seeds(&self, rid: &Id) -> Result { - match self.routing.get(rid) { - Ok(seeds) => { - Ok(seeds - .into_iter() - .fold(Seeds::new(self.rng.clone()), |mut seeds, node| { - if node != self.node_id() { - let addrs: Vec = self - .addresses - .get(&node) - .ok() - .flatten() - .map(|n| n.addrs) - .unwrap_or(vec![]); + let mut seeds = Seeds::new(self.rng.clone()); - if let Some(s) = self.sessions.get(&node) { - seeds.insert(Seed::new(node, addrs, Some(s.state.clone()))); - } else { - seeds.insert(Seed::new(node, addrs, None)); - } + // First build a list from peers that have synced our own refs, if any. + // This step is skipped if we don't have the repository yet, or don't have + // our own refs. + if let Ok(repo) = self.storage.repository(*rid) { + if let Ok(local) = RefsAt::new(&repo, self.node_id()) { + for seed in self.addresses.seeds(rid)? { + let seed = seed?; + let state = self.sessions.get(&seed.nid).map(|s| s.state.clone()); + let synced = if local.at == seed.synced_at.oid { + SyncStatus::Synced { at: seed.synced_at } + } else { + SyncStatus::OutOfSync { + local: local.at, + remote: seed.synced_at.oid, } - seeds - })) + }; + seeds.insert(Seed::new(seed.nid, seed.addresses, state, Some(synced))); + } } - Err(err) => Err(Error::Routing(err)), } + + // Then, add peers we know about but have no information about the sync status. + // These peers have announced that they track the repository via an inventory + // announcement, but we haven't received any ref announcements from them. + for nid in self.routing.get(rid)? { + if nid == self.node_id() { + continue; + } + if seeds.contains(&nid) { + // We already have a richer entry for this node. + continue; + } + let addrs = self.addresses.addresses(&nid)?; + let state = self.sessions.get(&nid).map(|s| s.state.clone()); + + seeds.insert(Seed::new(nid, addrs, state, None)); + } + Ok(seeds) } /// Return a new filter object, based on our tracking policy. diff --git a/radicle/src/node.rs b/radicle/src/node.rs index e224e3a1..6d1b426e 100644 --- a/radicle/src/node.rs +++ b/radicle/src/node.rs @@ -26,7 +26,7 @@ use crate::identity::Id; use crate::profile; use crate::storage::RefUpdate; -pub use address::KnownAddress; +pub use address::{KnownAddress, SyncedAt}; pub use config::Config; pub use cyphernet::addr::{HostName, PeerAddr}; pub use events::{Event, Events}; @@ -125,6 +125,26 @@ impl fmt::Display for State { } } +/// Repository sync status for our own refs. +#[derive(Debug, PartialEq, Eq, Clone, serde::Serialize, serde::Deserialize)] +#[serde(tag = "status")] +pub enum SyncStatus { + /// We're in sync. + #[serde(rename_all = "camelCase")] + Synced { + /// At what ref was the remote synced at. + at: SyncedAt, + }, + /// We're out of sync. + #[serde(rename_all = "camelCase")] + OutOfSync { + /// Local head of our `rad/sigrefs`. + local: git_ext::Oid, + /// Remote head of our `rad/sigrefs`. + remote: git_ext::Oid, + }, +} + /// Node alias. #[derive(Debug, PartialEq, Eq, Clone, serde::Serialize, serde::Deserialize)] pub struct Alias(String); @@ -427,6 +447,7 @@ pub struct Seed { pub nid: NodeId, pub addrs: Vec, pub state: Option, + pub sync: Option, } impl Seed { @@ -435,8 +456,18 @@ impl Seed { matches!(self.state, Some(State::Connected { .. })) } - pub fn new(nid: NodeId, addrs: Vec, state: Option) -> Self { - Self { nid, addrs, state } + pub fn new( + nid: NodeId, + addrs: Vec, + state: Option, + sync: Option, + ) -> Self { + Self { + nid, + addrs, + state, + sync, + } } } @@ -457,6 +488,11 @@ impl Seeds { self.0.insert(seed.nid, seed); } + /// Check membership. + pub fn contains(&self, nid: &NodeId) -> bool { + self.0.contains_key(nid) + } + /// Partitions the list of seeds into connected and disconnected seeds. /// Note that the disconnected seeds may be in a "connecting" state. pub fn partition(&self) -> (Vec, Vec) {