node: Use refs cache to decide on fetching — heartwood

When deciding whether or not to fetch a repo based on a refs announcement, we need to know whether our signed refs are stale or not, compared to the ones in the announcement. This can be quite expensive as it requires multiple reads from Git’s ODB, which often results in packfile loading.

To remedy this, we introduce a refs cache in our node database that is consulted when an announcement is received.

We also remove the logic that generates info messages, since they are rarely needed and introduced more complexity.

Additionally, we revert commit 3ad2b4431f4406fbb96c30d1edc77c5d969f17cb, which introduced logic to minimze wants and haves with additional logic, since it potentially also increases the load on the ODB.

 pub use transport::Transport;
 use radicle::crypto::PublicKey;
 use radicle::storage::refs::SignedRefsUpdate;
 use radicle::storage::refs::RefsAt;
 use radicle::storage::ReadRepository as _;
 use state::FetchState;
 use thiserror::Error;

     handle: &mut Handle<S>,
     limit: FetchLimit,
     remote: PublicKey,
     refs_at: Option<Vec<SignedRefsUpdate>>,
     refs_at: Option<Vec<RefsAt>>,
 ) -> Result<FetchResult, Error>
 where
     S: transport::ConnectionStream,

 use std::collections::{BTreeMap, BTreeSet};
 use std::ops::{Deref, DerefMut, Not as _};
 use std::ops::{Deref, Not as _};
 pub use radicle::storage::refs::{DiffedRefs, SignedRefsAt};
 pub use radicle::storage::refs::SignedRefsAt;
 pub use radicle::storage::{git::Validation, Validations};
 use radicle::storage::refs;
 use radicle::{crypto::PublicKey, storage::ValidateRepository};
 use crate::state::Cached;

     }
 }
 /// A set of [`DiffedRefs`] per remote `PublicKey`.
 ///
 /// To construct use [`DiffedRefs::load`].
 #[derive(Clone, Debug, Default)]
 pub struct RemoteDiffedRefs(BTreeMap<PublicKey, DiffedRefs>);
 impl RemoteDiffedRefs {
     /// Given a set of [`refs::RefsUpdate`]s, compute its
     /// [`DiffedRefs`] and use its [`refs::RefsUpdate::remote`] as the
     /// key for the `RemoteDiffedRefs` entry.
     ///
     /// If the `remote` is in the `may` set, then it is allowed to
     /// fail and will not be inserted in the set iff it does fail to
     /// load.
     ///
     /// If the `remote` is in the `must` set, then this method will
     /// fail iff loading the `DiffedRefs` fails.
     pub(crate) fn load<S>(
         cached: &Cached<S>,
         updates: Vec<refs::SignedRefsUpdate>,
         Select { must, may }: Select,
     ) -> Result<Self, error::Load> {
         updates
             .into_iter()
             .try_fold(Self::default(), |mut refs, update| {
                 match cached.load_diffed_refs(&update) {
                     Ok(diff) => {
                         refs.insert(update.remote, diff);
                         Ok(refs)
                     }
                     Err(e) if must.contains(&update.remote) => Err(e),
                     Err(_) if may.contains(&update.remote) => Ok(refs),
                     Err(e) => Err(e),
                 }
             })
     }
 }
 impl Deref for RemoteDiffedRefs {
     type Target = BTreeMap<PublicKey, DiffedRefs>;
     fn deref(&self) -> &Self::Target {
         &self.0
     }
 }
 impl DerefMut for RemoteDiffedRefs {
     fn deref_mut(&mut self) -> &mut Self::Target {
         &mut self.0
     }
 }
 impl<'a> IntoIterator for &'a RemoteDiffedRefs {
     type Item = <&'a BTreeMap<PublicKey, DiffedRefs> as IntoIterator>::Item;
     type IntoIter = <&'a BTreeMap<PublicKey, DiffedRefs> as IntoIterator>::IntoIter;
     fn into_iter(self) -> Self::IntoIter {
         self.0.iter()
     }
 }
 pub struct Select<'a> {
     pub must: &'a BTreeSet<PublicKey>,
     pub may: &'a BTreeSet<PublicKey>,

 //!      `rad/sigrefs`, for each configured namespace, i.e. followed
 //!      and delegate peers if the scope is "followed" and all peers is the
 //!      scope is all.
 //!   3. [`DataRefs`]/[`DiffedRefs`]: fetches the `Oid`s for each
 //!      reference listed in the `rad/sigrefs` for each fetched peer
 //!      in the [`SpecialRefs`] stage. Additionally, any references
 //!      that have been removed from `rad/sigrefs` are marked for
 //!      deletion.
 //!   3. [`DataRefs`]: fetches the `Oid`s for each reference listed in
 //!      the `rad/sigrefs` for each fetched peer in the
 //!      [`SpecialRefs`] stage. Additionally, any references that have
 //!      been removed from `rad/sigrefs` are marked for deletion.
 //!
 //! ### Pull
 //!
 //! A `pull` is split into two stages:
 //!
 //!   1. [`SpecialRefs`]: see above.
 //!   2. [`DataRefs`]/[`DiffedRefs`]: see above.
 //!   2. [`DataRefs`]: see above.
 use std::collections::{BTreeMap, BTreeSet, HashSet};

 }
 /// The [`ProtocolStage`] for fetching data refs from the set of
 /// `remotes`.
 /// remotes in `trusted`.
 ///
 /// All refs that are listed in the `remotes` sigrefs are checked
 /// against our refdb/odb to build a set of `wants` and `haves`. The

     }
 }
 /// The [`ProtocolStage`] that is similar to [`DataRefs`], however it
 /// is aware that it is an update of `rad/sigrefs`. This means that it
 /// will only compute `wants` and `haves` based on any modified
 /// `Oid`s.
 ///
 /// All refs and objects are prepared for updating as per usual, since
 /// we keep track of in-memory references for validation.
 #[derive(Debug)]
 pub struct DiffedRefs {
     /// The node that is being fetched from.
     pub remote: PublicKey,
     /// The set of signed references from each remote that was
     /// fetched.
     pub remotes: sigrefs::RemoteDiffedRefs,
     /// The data limit for this stage of fetching.
     pub limit: u64,
 }
 impl ProtocolStage for DiffedRefs {
     fn ls_refs(&self) -> Option<NonEmpty<BString>> {
         None
     }
     fn ref_filter(&self, _r: Ref) -> Option<ReceivedRef> {
         None
     }
     fn pre_validate(&self, _refs: &[ReceivedRef]) -> Result<(), error::Layout> {
         Ok(())
     }
     fn wants_haves(
         &self,
         refdb: &Repository,
         _refs: &[ReceivedRef],
     ) -> Result<WantsHaves, error::WantsHaves> {
         let mut wants_haves = WantsHaves::default();
         for (remote, refs) in &self.remotes {
             wants_haves.add(
                 refdb,
                 refs.iter().filter_map(|(refname, up)| {
                     let refname = Qualified::from_refstr(refname)
                         .map(|refname| refname.with_namespace(Component::from(remote)))?;
                     let tip = up.modified()?;
                     Some((refname, *tip))
                 }),
             )?;
         }
         Ok(wants_haves)
     }
     fn prepare_updates<'a>(
         &self,
         _s: &FetchState,
         _repo: &Repository,
         _refs: &'a [ReceivedRef],
     ) -> Result<Updates<'a>, error::Prepare> {
         use radicle::storage::refs::Update::{Added, Changed, Deleted, Same};
         let mut updates = Updates::default();
         let prefix_rad = refname!("refs/rad");
         for (remote, refs) in &self.remotes {
             for (name, up) in refs.iter() {
                 let is_refs_rad = name.starts_with(prefix_rad.as_str());
                 let tracking: Namespaced<'_> = Qualified::from_refstr(name)
                     .and_then(|q| refs::ReceivedRefname::remote(*remote, q).to_namespaced())
                     .expect("we checked sigrefs well-formedness in wants_refs already");
                 match up {
                     Added { oid } | Changed { oid } => updates.add(
                         *remote,
                         Update::Direct {
                             name: tracking,
                             target: *oid,
                             no_ff: Policy::Allow,
                         },
                     ),
                     Deleted { oid } if !is_refs_rad => updates.add(
                         *remote,
                         Update::Prune {
                             name: tracking,
                             prev: either::Left(*oid),
                         },
                     ),
                     // N.b. create an update for this reference so
                     // that the in-memory refdb is updated.
                     Same { oid } => updates.add(
                         *remote,
                         Update::Direct {
                             name: tracking,
                             target: *oid,
                             no_ff: Policy::Allow,
                         },
                     ),
                     // N.b. `refs/rad` is not subject to pruning.
                     Deleted { .. } => continue,
                 }
             }
         }
         Ok(updates)
     }
 }
 // N.b. the `delegates` are the delegates of the repository, with the
 // potential removal of the local peer in the case of a `pull`.
 fn special_refs_updates<'a>(

 use radicle::prelude::Verified;
 use radicle::storage;
 use radicle::storage::refs::{RefsAt, SignedRefs, SignedRefsUpdate};
 use radicle::storage::refs::{RefsAt, SignedRefs};
 use radicle::storage::{
     git::Validation, Remote, RemoteId, RemoteRepository, Remotes, ValidateRepository, Validations,
 };

         Ok(fetched)
     }
     /// Fetch the set of special refs, depending on `refs_updates`.
     /// Fetch the set of special refs, depending on `refs_at`.
     ///
     /// If `refs_updates` is `Some`, then run the [`SigrefsAt`] stage,
     /// If `refs_at` is `Some`, then run the [`SigrefsAt`] stage,
     /// which specifically fetches `rad/sigrefs` which are listed in
     /// `refs_at`. It then runs the [`DiffedRefs`] stage to fetch the
     /// difference between old references listed existing
     /// `rad/sigrefs` and the new `rad/sigrefs` fetched in the
     /// previous step.
     /// `refs_at`.
     ///
     /// If `refs_updates` is `None`, then run the [`SpecialRefs`]
     /// stage, which fetches `rad/sigrefs` and `rad/id` from all
     /// tracked and delegate peers (scope dependent). It then runs the
     /// [`DataRefs`] stage to fetch the references listed in the
     /// `rad/sigrefs` fetched in the previous step.
     /// If `refs_at` is `None`, then run the [`SpecialRefs`] stage,
     /// which fetches `rad/sigrefs` and `rad/id` from all tracked and
     /// delegate peers (scope dependent).
     ///
     /// The resulting `Vec<PublicKey>` will be the set of fetched
     /// remotes.
     /// The resulting [`sigrefs::RemoteRefs`] will be the set of
     /// `rad/sigrefs` of the fetched remotes.
     fn run_special_refs<S>(
         &mut self,
         handle: &mut Handle<S>,

         delegates: BTreeSet<PublicKey>,
         limit: &FetchLimit,
         remote: PublicKey,
         refs_updates: Option<Vec<SignedRefsUpdate>>,
     ) -> Result<Vec<PublicKey>, error::Protocol>
         refs_at: Option<Vec<RefsAt>>,
     ) -> Result<sigrefs::RemoteRefs, error::Protocol>
     where
         S: transport::ConnectionStream,
     {
         match refs_updates {
             Some(refs_updates) => {
                 let (must, may): (BTreeSet<PublicKey>, BTreeSet<PublicKey>) = refs_updates
         match refs_at {
             Some(refs_at) => {
                 let (must, may): (BTreeSet<PublicKey>, BTreeSet<PublicKey>) = refs_at
                     .iter()
                     .map(|refs_at| refs_at.remote)
                     .partition(|id| delegates.contains(id));
                 let refs_at = refs_updates.iter().map(RefsAt::from).collect();
                 let sigrefs_at = stage::SigrefsAt {
                     remote,
                     delegates,

                 log::trace!(target: "fetch", "{sigrefs_at:?}");
                 self.run_stage(handle, handshake, &sigrefs_at)?;
                 let remotes = sigrefs::RemoteDiffedRefs::load(
                 let signed_refs = sigrefs::RemoteRefs::load(
                     &self.as_cached(handle),
                     refs_updates,
                     sigrefs::Select {
                         must: &must,
                         may: &may,
                     },
                 )?;
                 let fetched = remotes.keys().copied().collect();
                 let refs = stage::DiffedRefs {
                     remote,
                     remotes,
                     limit: limit.refs,
                 };
                 self.run_stage(handle, handshake, &refs)?;
                 Ok(fetched)
                 Ok(signed_refs)
             }
             None => {
                 let followed = handle.allowed();

                 log::trace!(target: "fetch", "{special_refs:?}");
                 let fetched = self.run_stage(handle, handshake, &special_refs)?;
                 let remotes = sigrefs::RemoteRefs::load(
                 let signed_refs = sigrefs::RemoteRefs::load(
                     &self.as_cached(handle),
                     sigrefs::Select {
                         must: &delegates,

         log::trace!(target: "fetch", "Identity delegates {delegates:?}");
         let fetched = self.run_special_refs(
         let signed_refs = self.run_special_refs(
             handle,
             handshake,
             delegates.clone(),

         )?;
         log::debug!(
             target: "fetch",
             "Fetched data refs for {} remote(s) ({}ms)",
             fetched.len(),
             "Fetched data for {} remote(s) ({}ms)",
             signed_refs.len(),
             start.elapsed().as_millis()
         );
         let data_refs = stage::DataRefs {
             remote,
             remotes: signed_refs,
             limit: limit.refs,
         };
         self.run_stage(handle, handshake, &data_refs)?;
         log::debug!(
             target: "fetch",
             "Fetched data refs for {} remotes ({}ms)",
             data_refs.remotes.len(),
             start.elapsed().as_millis()
         );

         // added to `warnings`.
         let mut warnings = sigrefs::Validations::default();
         let mut failures = sigrefs::Validations::default();
         let signed_refs = data_refs.remotes;
         // We may prune fetched remotes, so we keep track of
         // non-pruned, fetched remotes here.

         // TODO(finto): this might read better if it got its own
         // private function.
         for remote in &fetched {
         for remote in signed_refs.keys() {
             if handle.is_blocked(remote) {
                 continue;
             }

         }
     }
     pub fn load_diffed_refs(
         &self,
         update: &SignedRefsUpdate,
     ) -> Result<storage::refs::DiffedRefs, sigrefs::error::Load> {
         update.difference(&self.handle.repo)
     }
     #[allow(dead_code)]
     pub(crate) fn inspect(&self) {
         self.state.refs.inspect()

         let policy = config.policy;
         log::info!(target: "node", "Opening node database..");
         let mut db: service::Stores<_> = home.database_mut()?.into();
         let db = home.database_mut()?;
         let mut stores: service::Stores<_> = db.clone().into();
         log::info!(target: "node", "Opening policy database..");
         let policies = home.policies_mut()?;

                 .expect("Runtime::init: unable to solve proof-of-work puzzle")
         };
         if config.connect.is_empty() && db.addresses().is_empty()? {
         if config.connect.is_empty() && stores.addresses().is_empty()? {
             log::info!(target: "node", "Address book is empty. Adding bootstrap nodes..");
             for (alias, addr) in config.network.bootstrap() {
                 let (id, addr) = addr.into();
                 db.addresses_mut().insert(
                 stores.addresses_mut().insert(
                     &id,
                     radicle::node::Features::SEED,
                     alias,

                     [node::KnownAddress::new(addr, address::Source::Bootstrap)],
                 )?;
             }
             log::info!(target: "node", "{} nodes added to address book", db.addresses().len()?);
             log::info!(target: "node", "{} nodes added to address book", stores.addresses().len()?);
         }
         let emitter: Emitter<Event> = Default::default();
         let mut service = service::Service::new(
             config.clone(),
             clock,
             db,
             stores,
             storage.clone(),
             policies,
             signer.clone(),

             handle.clone(),
             notifications,
             cobs_cache,
             db,
             worker::Config {
                 capacity: config.workers,
                 storage: storage.clone(),

 use radicle::node::address::Store as _;
 use radicle::node::address::{AddressBook, KnownAddress};
 use radicle::node::config::PeerConfig;
 use radicle::node::refs::Store as _;
 use radicle::node::routing::Store as _;
 use radicle::node::seed;
 use radicle::node::seed::Store as _;
 use radicle::node::{ConnectOptions, Penalty, Severity};
 use radicle::storage::refs::SignedRefsUpdate;
 use radicle::storage::refs::SIGREFS_BRANCH;
 use radicle::storage::{Inventory, RepositoryError};
 use crate::crypto;

 use crate::prelude::*;
 use crate::runtime::Emitter;
 use crate::service::gossip::Store as _;
 use crate::service::message::{Announcement, AnnouncementMessage, Info, Ping};
 use crate::service::message::{NodeAnnouncement, RefsAnnouncement};
 use crate::service::message::{
     Announcement, AnnouncementMessage, Info, NodeAnnouncement, Ping, RefsAnnouncement, RefsStatus,
 };
 use crate::service::policy::{store::Write, Policy, Scope};
 use crate::storage;
 use crate::storage::refs::RefsAt;
 use crate::storage::ReadRepository;
 use crate::storage::{Namespaces, ReadStorage};
 use crate::storage::{refs::RefsAt, Namespaces, ReadRepository, ReadStorage};
 use crate::worker::fetch;
 use crate::worker::FetchError;
 use crate::Link;

 use self::io::Outbox;
 use self::limitter::RateLimiter;
 use self::message::{InventoryAnnouncement, RefsStatus};
 use self::message::InventoryAnnouncement;
 use self::policy::NamespacesError;
 /// How often to run the "idle" task.

 }
 /// A store for all node data.
 pub trait Store: address::Store + gossip::Store + routing::Store + seed::Store {}
 pub trait Store:
     address::Store + gossip::Store + routing::Store + seed::Store + node::refs::Store
 {
 }
 impl Store for node::Database {}

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

     /// Peer being fetched from.
     from: NodeId,
     /// Refs being fetched.
     refs_at: Vec<SignedRefsUpdate>,
     refs_at: Vec<RefsAt>,
     /// Result channel.
     channel: Option<chan::Sender<FetchResult>>,
 }

     pub fn seeds_mut(&mut self) -> &mut impl seed::Store {
         &mut self.0
     }
     /// Get the database as a refs db.
     pub fn refs(&self) -> &impl node::refs::Store {
         &self.0
     }
     /// Get the database as a refs db, mutably.
     pub fn refs_mut(&mut self) -> &mut impl node::refs::Store {
         &mut self.0
     }
 }
 impl<D> From<D> for Stores<D> {

                     .expect("Service::command: error unfollowing node");
                 resp.send(updated).ok();
             }
             Command::AnnounceRefs(id, resp) => match self.announce_refs(id, [self.node_id()]) {
             Command::AnnounceRefs(id, resp) => match self.announce_own_refs(id) {
                 Ok(refs) => match refs.as_slice() {
                     &[refs] => {
                         resp.send(refs).ok();

         &mut self,
         rid: RepoId,
         from: NodeId,
         refs: NonEmpty<SignedRefsUpdate>,
         refs: NonEmpty<RefsAt>,
         timeout: time::Duration,
         channel: Option<chan::Sender<FetchResult>>,
     ) {

                 .seed_policy(&rid)
                 .expect("Service::dequeue_fetch: error accessing repo seeding configuration");
             let refs_at = refs_at.iter().map(RefsAt::from).collect();
             match self.refs_status_of(rid, refs_at, &repo_entry.scope) {
                 Ok(status) => {
                     if let Some(refs) = NonEmpty::from_vec(status.fresh) {

                         info!(target: "service", "Routing table updated for {} with seed {announcer}", message.rid);
                     }
                 }
                 let Some(refs) = NonEmpty::from_vec(message.refs.to_vec()) else {
                     debug!(target: "service", "Skipping fetch, empty refs announcement for {}", message.rid);
                     return Ok(false);
                 };
                 // Update sync status for this repo.
                 // Update sync status of announcer for this repo.
                 if let Some(refs) = message.refs.iter().find(|r| &r.remote == self.nid()) {
                     match self.db.seeds_mut().synced(
                         &message.rid,

                 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.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}");
                             return Ok(relay);
                         }
                     };
                     // Refs can be relayed by peers who don't have the data in storage,
                     // therefore we only check whether we are connected to the *announcer*,
                     // which is required by the protocol to only announce refs it has.
                     if let Some(remote) = self.sessions.get(announcer).cloned() {
                         // If the relayer is also the origin of the message, we inform it
                         // about any refs that are already in sync (stale).
                         if relayer == announcer {
                             // If the stale refs contain refs announced by the peer, let it know
                             // that we're already in sync.
                             if let Some(at) = stale
                                 .iter()
                                 .find(|refs| refs.remote == remote.id)
                                 .copied()
                                 .map(|RefsAt { at, .. }| at)
                             {
                                 debug!(
                                     target: "service", "Refs of {} already synced for {} at {at}",
                                     remote.id,
                                     message.rid,
                                 );
                                 self.outbox.write(
                                     &remote,
                                     Info::RefsAlreadySynced {
                                         rid: message.rid,
                                         at,
                                     }
                                     .into(),
                                 );
                             }
                         }
                         // 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, None);
                         } else {
                             debug!(target: "service", "Skipping fetch, all refs of {} are already in local storage", message.rid);
                         }
                     } else {
                         trace!(
                             target: "service",
                             "Skipping fetch of {}, no sessions connected to {announcer}",
                             message.rid
                         );
                     }
                     return Ok(relay);
                 } else {
                 if repo_entry.policy != Policy::Allow {
                     debug!(
                         target: "service",
                         "Ignoring refs announcement from {announcer}: repository {} isn't seeded",
                         message.rid
                     );
                     return Ok(false);
                 }
                 // Refs can be relayed by peers who don't have the data in storage,
                 // therefore we only check whether we are connected to the *announcer*,
                 // which is required by the protocol to only announce refs it has.
                 let Some(remote) = self.sessions.get(announcer).cloned() else {
                     trace!(
                         target: "service",
                         "Skipping fetch of {}, no sessions connected to {announcer}",
                         message.rid
                     );
                     return Ok(relay);
                 };
                 // Finally, if there's anything to fetch, we fetch it from the remote.
                 self.fetch_refs_at(message.rid, remote.id, refs, FETCH_TIMEOUT, None);
                 return Ok(relay);
             }
             AnnouncementMessage::Node(
                 ann @ NodeAnnouncement {

     pub fn handle_info(&mut self, remote: NodeId, info: &Info) -> Result<(), session::Error> {
         match info {
             // Nb. We don't currently send this message.
             Info::RefsAlreadySynced { rid, at } => {
                 debug!(target: "service", "Refs already synced for {rid} by {remote}");
                 self.emitter.emit(Event::RefsSynced {

         Ok(())
     }
     /// A convenient method to check if we should fetch from a `RefsAnnouncement` with `scope`.
     fn refs_status_of(
         &self,
         rid: RepoId,
         refs: Vec<RefsAt>,
         scope: &policy::Scope,
     ) -> Result<RefsStatus, Error> {
         let mut refs = RefsStatus::new(rid, refs, &self.storage)?;
         // First, check the freshness.
         if refs.fresh.is_empty() {
             return Ok(refs);
         }
         // Second, check the scope.
         match scope {
             policy::Scope::All => Ok(refs),
             policy::Scope::Followed => {
                 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.
                         followed.remove(self.nid());
                         refs.fresh.retain(|r| followed.contains(&r.remote));
                         Ok(refs)
                     }
                     Err(e) => Err(e.into()),
                 }
             }
         }
     }
     pub fn handle_message(
         &mut self,
         remote: &NodeId,