93a1ac7 node: Allow announcing refs for given public key — heartwood

modified crates/radicle-cli/src/node.rs

@@ -199,18 +199,29 @@ where

         reporting.completion.clone(),
     );
     match node.announce(rid, settings.timeout, announcer, |node, progress| {
         spinner.message(format!(
             "Synced with {}, {} of {} preferred seeds, and {} of at least {} replica(s).",
             term::format::node_id_human_compact(node),
             term::format::secondary(progress.preferred()),
             term::format::secondary(n_preferred_seeds),
             term::format::secondary(progress.synced()),
             // N.b. the number of replicas could exceed the target if we're
             // waiting for preferred seeds
             term::format::secondary(min_replicas.max(progress.synced())),
         ));
     }) {
     // Note that technically we could command the node to announce refs
     // for an arbitrary namespace. Here, for backwards compatibility, we
     // only announce for our own namespace.
     let ns = radicle::crypto::PublicKey::from(profile.did());
     match node.announce(
         rid,
         Some(ns),
         settings.timeout,
         announcer,
         |node, progress| {
             spinner.message(format!(
                 "Synced with {}, {} of {} preferred seeds, and {} of at least {} replica(s).",
                 term::format::node_id_human_compact(node),
                 term::format::secondary(progress.preferred()),
                 term::format::secondary(n_preferred_seeds),
                 term::format::secondary(progress.synced()),
                 // N.b. the number of replicas could exceed the target if we're
                 // waiting for preferred seeds
                 term::format::secondary(min_replicas.max(progress.synced())),
             ));
         },
     ) {
         Ok(result) => {
             spinner.message(format!(
                 "Synced with {} seed(s)",

modified crates/radicle-cli/tests/commands.rs

@@ -1772,7 +1772,7 @@ fn test_cob_replication() {

     // announcement, otherwise Alice will consider it stale.
     thread::sleep(time::Duration::from_millis(3));
     bob.handle.announce_refs(rid).unwrap();
     bob.handle.announce_refs(rid, None).unwrap();
     // Wait for Alice to fetch the issue refs.
     events

modified crates/radicle-node/src/control.rs

@@ -219,8 +219,8 @@ where

                 return Err(CommandError::Runtime(e));
             }
         },
         Command::AnnounceRefs { rid } => {
             let refs = handle.announce_refs(rid)?;
         Command::AnnounceRefs { rid, ns } => {
             let refs = handle.announce_refs(rid, ns)?;
             CommandResult::Okay(refs).to_writer(writer)?;
         }

@@ -308,6 +308,7 @@ mod tests {

         let socket = tmp.path().join("alice.sock");
         let rids = test::arbitrary::set::<RepoId>(1..3);
         let listener = Listener::bind(&socket).unwrap();
         let nid = handle.nid().unwrap();
         thread::spawn({
             let handle = handle.clone();

@@ -325,7 +326,8 @@ mod tests {

                 &mut stream,
                 "{}",
                 json::to_string(&Command::AnnounceRefs {
                     rid: rid.to_owned()
                     rid: rid.to_owned(),
                     ns: None,
                 })
                 .unwrap()
             )

@@ -345,7 +347,7 @@ mod tests {

         }
         for rid in &rids {
             assert!(handle.updates.lock().unwrap().contains(rid));
             assert!(handle.updates.lock().unwrap().contains(&(*rid, nid)));
         }
     }

modified crates/radicle-node/src/runtime/handle.rs

@@ -9,6 +9,7 @@ use std::os::unix::net::UnixStream as Stream;

 use winpipe::WinStream as Stream;
 use crossbeam_channel as chan;
 use radicle::crypto::PublicKey;
 use radicle::node::events::{Event, Events};
 use radicle::node::policy;
 use radicle::node::{Config, NodeId};

@@ -253,9 +254,9 @@ impl radicle::node::Handle for Handle {

         receiver.recv().map_err(Error::from)
     }
     fn announce_refs(&mut self, id: RepoId) -> Result<RefsAt, Error> {
     fn announce_refs(&mut self, id: RepoId, ns: Option<PublicKey>) -> Result<RefsAt, Error> {
         let (sender, receiver) = chan::bounded(1);
         self.command(service::Command::AnnounceRefs(id, sender))?;
         self.command(service::Command::AnnounceRefs(id, ns, sender))?;
         receiver.recv().map_err(Error::from)
     }

modified crates/radicle-node/src/test/handle.rs

@@ -3,6 +3,7 @@ use std::str::FromStr;

 use std::sync::{Arc, Mutex};
 use std::time;
 use radicle::crypto::PublicKey;
 use radicle::git;
 use radicle::storage::refs::RefsAt;

@@ -14,7 +15,7 @@ use radicle::node::NodeId;

 #[derive(Default, Clone)]
 pub struct Handle {
     pub updates: Arc<Mutex<Vec<RepoId>>>,
     pub updates: Arc<Mutex<Vec<(RepoId, PublicKey)>>>,
     pub seeding: Arc<Mutex<HashSet<RepoId>>>,
     pub following: Arc<Mutex<HashSet<NodeId>>>,
 }

@@ -91,8 +92,11 @@ impl radicle::node::Handle for Handle {

         Ok(self.following.lock().unwrap().remove(&id))
     }
     fn announce_refs(&mut self, id: RepoId) -> Result<RefsAt, Self::Error> {
         self.updates.lock().unwrap().push(id);
     fn announce_refs(&mut self, id: RepoId, ns: Option<PublicKey>) -> Result<RefsAt, Self::Error> {
         self.updates
             .lock()
             .unwrap()
             .push((id, ns.unwrap_or_else(|| self.nid().unwrap())));
         Ok(RefsAt {
             remote: self.nid()?,

modified crates/radicle-node/src/tests/e2e.rs

@@ -1378,7 +1378,7 @@ fn test_background_foreground_fetch() {

         Title::new("Concurrent fetches").unwrap(),
         "Concurrent fetches are harshing my vibes",
     );
     bob.handle.announce_refs(rid).unwrap();
     bob.handle.announce_refs(rid, None).unwrap();
     alice_events
         .wait(
             |e| matches!(e, Event::RefsAnnounced { .. }).then_some(()),

@@ -1427,7 +1427,7 @@ fn test_catchup_on_refs_announcements() {

     log::debug!(target: "test", "Bob creating his issue..");
     bob.issue(acme, Title::new("Bob's issue").unwrap(), "[..]");
     bob.handle.announce_refs(acme).unwrap();
     bob.handle.announce_refs(acme, None).unwrap();
     log::debug!(target: "test", "Waiting for seed to fetch Bob's refs from Bob..");
     seed.has_remote_of(&acme, &bob.id); // Seed fetches Bob's refs.

modified crates/radicle-protocol/src/service.rs

@@ -238,8 +238,8 @@ pub type QueryState = dyn Fn(&dyn ServiceState) -> Result<(), CommandError> + Se

 /// Commands sent to the service by the operator.
 pub enum Command {
     /// Announce repository references for given repository to peers.
     AnnounceRefs(RepoId, chan::Sender<RefsAt>),
     /// Announce repository references for given repository and namespace to peers.
     AnnounceRefs(RepoId, Option<PublicKey>, chan::Sender<RefsAt>),
     /// Announce local repositories to peers.
     AnnounceInventory,
     /// Add repository to local inventory.

@@ -271,7 +271,7 @@ pub enum Command {

 impl fmt::Debug for Command {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
             Self::AnnounceRefs(id, _) => write!(f, "AnnounceRefs({id})"),
             Self::AnnounceRefs(id, _, _) => write!(f, "AnnounceRefs({id})"),
             Self::AnnounceInventory => write!(f, "AnnounceInventory"),
             Self::AddInventory(rid, _) => write!(f, "AddInventory({rid})"),
             Self::Connect(id, addr, opts) => write!(f, "Connect({id}, {addr}, {opts:?})"),

@@ -930,7 +930,7 @@ where

                     .expect("Service::command: error unfollowing node");
                 resp.send(updated).ok();
             }
             Command::AnnounceRefs(id, resp) => {
             Command::AnnounceRefs(id, ns, resp) => {
                 let doc = match self.storage.get(id) {
                     Ok(Some(doc)) => doc,
                     Ok(None) => {

@@ -943,8 +943,13 @@ where

                     }
                 };
                 match self.announce_own_refs(id, doc) {
                     Ok(refs) => match refs.as_slice() {
                 let result = match ns {
                     Some(ns) => self.announce_refs(id, doc, [ns]),
                     None => self.announce_own_refs(id, doc),
                 };
                 match result {
                     Ok((refs, _timestamp)) => match refs.as_slice() {
                         &[refs] => {
                             resp.send(refs).ok();
                         }

@@ -2167,7 +2172,11 @@ where

     }
     /// Announce our own refs for the given repo.
     fn announce_own_refs(&mut self, rid: RepoId, doc: Doc) -> Result<Vec<RefsAt>, Error> {
     fn announce_own_refs(
         &mut self,
         rid: RepoId,
         doc: Doc,
     ) -> Result<(Vec<RefsAt>, Timestamp), Error> {
         let (refs, timestamp) = self.announce_refs(rid, doc, [self.node_id()])?;
         // Update refs database with our signed refs branches.

@@ -2193,7 +2202,7 @@ where

                 );
             }
         }
         Ok(refs)
         Ok((refs, timestamp))
     }
     /// Announce local refs for given repo.

modified crates/radicle/src/node.rs

@@ -613,9 +613,28 @@ impl From<Address> for HostName {

 #[serde(rename_all = "camelCase", tag = "command")]
 #[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
 pub enum Command {
     /// Announce repository references for given repository to peers.
     /// Announce repository references for given repository
     /// and namespace to peers.
     #[serde(rename_all = "camelCase")]
     AnnounceRefs { rid: RepoId },
     AnnounceRefs {
         rid: RepoId,
         /// The namespace for which references should be announced.
         ///
         /// For backwards compatibility this is optional and
         /// omission is interpreted by the node as if the Node ID
         /// of the node itself, i.e., the node that this command is
         /// received by, was passed. Thus, the node would announce
         /// "it's own" references. This makes perfect sense when
         /// the node and the user have the same cryptographic identity
         /// but not when they are different.
         #[cfg_attr(
             feature = "schemars",
             schemars(with = "Option<crate::schemars_ext::crypto::PublicKey>")
         )]
         #[serde(default, skip_serializing_if = "Option::is_none")]
         ns: Option<PublicKey>,
     },
     /// Announce local repositories to peers.
     #[serde(rename_all = "camelCase")]

@@ -624,7 +643,7 @@ pub enum Command {

     /// Update node's inventory.
     AddInventory { rid: RepoId },
     /// Get the current node condiguration.
     /// Get the current node configuration.
     Config,
     /// Get the node's listen addresses.

@@ -1113,7 +1132,7 @@ pub trait Handle: Clone + Sync + Send {

     /// Unfollow the given peer.
     fn unfollow(&mut self, id: NodeId) -> Result<bool, Self::Error>;
     /// Notify the service that a project has been updated, and announce local refs.
     fn announce_refs(&mut self, id: RepoId) -> Result<RefsAt, Self::Error>;
     fn announce_refs(&mut self, id: RepoId, ns: Option<PublicKey>) -> Result<RefsAt, Self::Error>;
     /// Announce local inventory.
     fn announce_inventory(&mut self) -> Result<(), Self::Error>;
     /// Notify the service that our inventory was updated with the given repository.

@@ -1215,12 +1234,13 @@ impl Node {

     pub fn announce(
         &mut self,
         rid: RepoId,
         ns: Option<PublicKey>,
         timeout: time::Duration,
         mut announcer: sync::Announcer,
         mut report: impl FnMut(&NodeId, sync::announce::Progress),
     ) -> Result<sync::AnnouncerResult, Error> {
         let mut events = self.subscribe(timeout)?;
         let refs = self.announce_refs(rid)?;
         let refs = self.announce_refs(rid, ns)?;
         let started = time::Instant::now();

@@ -1391,9 +1411,9 @@ impl Handle for Node {

         Ok(response.updated)
     }
     fn announce_refs(&mut self, rid: RepoId) -> Result<RefsAt, Error> {
     fn announce_refs(&mut self, rid: RepoId, ns: Option<PublicKey>) -> Result<RefsAt, Error> {
         let refs: RefsAt = self
             .call(Command::AnnounceRefs { rid }, DEFAULT_TIMEOUT)?
             .call(Command::AnnounceRefs { rid, ns }, DEFAULT_TIMEOUT)?
             .next()
             .ok_or(Error::EmptyResponse)??;