node: Generalize commands announce/seeds to take public key — heartwood

The commands “announce references” and “seeds” only take a repository ID, and carry the implicit assumption that the namespace with the same public key as the Node ID of the receiving radicle-node process is relevant, but no other namespace.

In the spirit of separating user and node identity, relax this, so that these command also carry the public key for which the announcement should be made or sync status should be reported.

This feature is not exposed via CLI, but used to pass through the user identity of the active profile.

For sync status, this patch still is not fully satisfactory, because the node does not store sync updates for all namespaces, but only for its own. Storing sync updates for namespaces that were announced from the node could be implemented in the future.

     const SYMBOL_STATE_UNKNOWN: &str = "•";
     let mut table = Table::<5, term::Label>::new(TableOptions::bordered());
     let mut seeds: Vec<_> = node.seeds(rid)?.into();
     let mut seeds: Vec<_> = node.seeds_for(rid, [*profile.did()])?.into();
     let local_nid = node.nid()?;
     let aliases = profile.aliases();

         None => {
             // We push nodes that are in our seed list in attempt to fulfill the
             // replicas, if needed.
             let seeds = node.seeds(rid)?;
             let seeds = node.seeds_for(rid, [*profile.did()])?;
             let (connected, disconnected) = seeds.partition();
             let candidates = connected
                 .into_iter()

     let config = match sync::PrivateNetwork::private_repo(&doc) {
         None => {
             let (synced, unsynced) = node.seeds(rid)?.iter().fold(
             let (synced, unsynced) = node.seeds_for(rid, [*me])?.iter().fold(
                 (BTreeSet::new(), BTreeSet::new()),
                 |(mut synced, mut unsynced), seed| {
                     if seed.is_synced() {

         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())),
         ));
     }) {
     match node.announce(
         rid,
         [profile.did().into()],
         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)",

     let working = environment.tempdir().join("working");
     let rid = alice.project("heartwood", "Radicle Heartwood Protocol & Stack");
     let mut alice = alice.spawn();
     let seeds = alice.handle.seeds(rid).unwrap();
     let seeds = alice.handle.seeds_for(rid, [alice.id]).unwrap();
     let connected = seeds.connected().collect::<Vec<_>>();
     assert!(connected.is_empty());

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

             CommandResult::Okay(addrs).to_writer(writer)?;
         }
         #[allow(deprecated)]
         Command::Seeds { rid } => {
             let seeds = handle.seeds(rid)?;
             CommandResult::Okay(seeds).to_writer(writer)?;
         }
         Command::SeedsFor { rid, namespaces } => {
             let seeds = handle.seeds_for(rid, namespaces)?;
             CommandResult::Okay(seeds).to_writer(writer)?;
         }
         Command::Sessions => {
             let sessions = handle.sessions()?;

                 return Err(CommandError::Runtime(e));
             }
         },
         #[allow(deprecated)]
         Command::AnnounceRefs { rid } => {
             let refs = handle.announce_refs(rid)?;
             CommandResult::Okay(refs).to_writer(writer)?;
         }
         Command::AnnounceRefsFor { rid, namespaces } => {
             let refs = handle.announce_refs_for(rid, namespaces)?;
             CommandResult::Okay(refs).to_writer(writer)?;
         }
         Command::AnnounceInventory => {
             if let Err(e) = handle.announce_inventory() {
                 return Err(CommandError::Runtime(e));

         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();

             writeln!(
                 &mut stream,
                 "{}",
                 json::to_string(&Command::AnnounceRefs {
                     rid: rid.to_owned()
                 json::to_string(&Command::AnnounceRefsFor {
                     rid: rid.to_owned(),
                     namespaces: [nid].into(),
                 })
                 .unwrap()
             )

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

 use std::collections::HashSet;
 use std::net;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Arc;

 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};

             .map_err(Error::from)
     }
     fn seeds(&mut self, id: RepoId) -> Result<Seeds, Self::Error> {
     fn seeds_for(
         &mut self,
         id: RepoId,
         namespaces: impl IntoIterator<Item = PublicKey>,
     ) -> Result<Seeds, Self::Error> {
         let (sender, receiver) = chan::bounded(1);
         self.command(service::Command::Seeds(id, sender))?;
         self.command(service::Command::Seeds(
             id,
             HashSet::from_iter(namespaces),
             sender,
         ))?;
         receiver.recv().map_err(Error::from)
     }

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

 #[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>>>,
 }

         unimplemented!();
     }
     fn seeds(&mut self, _id: RepoId) -> Result<Seeds, Self::Error> {
     fn seeds_for(
         &mut self,
         _id: RepoId,
         _namespaces: impl IntoIterator<Item = PublicKey>,
     ) -> Result<Seeds, Self::Error> {
         unimplemented!();
     }

         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_for(
         &mut self,
         id: RepoId,
         namespaces: impl IntoIterator<Item = PublicKey>,
     ) -> Result<RefsAt, Self::Error> {
         self.updates
             .lock()
             .unwrap()
             .extend(namespaces.into_iter().map(|ns| (id, ns)));
         Ok(RefsAt {
             remote: self.nid()?,

         log::debug!(target: "test", "Waiting for {} to be in sync with {nid} for {rid}", self.id);
         loop {
             let seeds = self.handle.seeds(*rid).unwrap();
             let seeds = self.handle.seeds_for(*rid, [self.id]).unwrap();
             if seeds.iter().any(|s| s.nid == *nid && s.is_synced()) {
                 break;
             }

     let updated = alice.handle.seed(acme, Scope::All).unwrap();
     assert!(updated);
     let seeds = alice.handle.seeds(acme).unwrap();
     let seeds = alice.handle.seeds_for(acme, None).unwrap();
     assert!(seeds.is_connected(&bob.id));
     let result = alice.handle.fetch(acme, bob.id, DEFAULT_TIMEOUT).unwrap();

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

     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_for(acme, [bob.id]).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.

 /// 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 namespaces to peers.
     AnnounceRefs(RepoId, HashSet<PublicKey>, chan::Sender<RefsAt>),
     /// Announce local repositories to peers.
     AnnounceInventory,
     /// Add repository to local inventory.

     Config(chan::Sender<Config>),
     /// Get the node's listen addresses.
     ListenAddrs(chan::Sender<Vec<std::net::SocketAddr>>),
     /// Lookup seeds for the given repository in the routing table.
     Seeds(RepoId, chan::Sender<Seeds>),
     /// Lookup seeds for the given repository in the routing table, and report
     /// sync status for given namespaces.
     Seeds(RepoId, HashSet<PublicKey>, chan::Sender<Seeds>),
     /// Fetch the given repository from the network.
     Fetch(RepoId, NodeId, time::Duration, chan::Sender<FetchResult>),
     /// Seed the given repository.

 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:?})"),
             Self::Disconnect(id) => write!(f, "Disconnect({id})"),
             Self::Config(_) => write!(f, "Config"),
             Self::ListenAddrs(_) => write!(f, "ListenAddrs"),
             Self::Seeds(id, _) => write!(f, "Seeds({id})"),
             Self::Seeds(id, _, _) => write!(f, "Seeds({id})"),
             Self::Fetch(id, node, _, _) => write!(f, "Fetch({id}, {node})"),
             Self::Seed(id, scope, _) => write!(f, "Seed({id}, {scope})"),
             Self::Unseed(id, _) => write!(f, "Unseed({id})"),

             Command::ListenAddrs(resp) => {
                 resp.send(self.listening.clone()).ok();
             }
             Command::Seeds(rid, resp) => match self.seeds(&rid) {
             Command::Seeds(rid, namespaces, resp) => match self.seeds(&rid, namespaces) {
                 Ok(seeds) => {
                     let (connected, disconnected) = seeds.partition();
                     debug!(

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

                     }
                 };
                 match self.announce_own_refs(id, doc) {
                     Ok(refs) => match refs.as_slice() {
                         &[refs] => {
                             resp.send(refs).ok();
                 match self.announce_own_refs(id, doc, namespaces) {
                     Ok((refs, _timestamp)) => {
                         for r in refs {
                             resp.send(r).ok();
                         }
                         // SAFETY: Since we passed in one NID, we should get exactly one item back.
                         [..] => panic!("Service::command: unexpected refs returned"),
                     },
                     }
                     Err(err) => {
                         error!(target: "service", "Error announcing refs: {err}");
                     }

                 } 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) {
                     if let Err(e) = self.announce_refs(rid, doc.into(), namespaces, false) {
                         error!(target: "service", "Failed to announce new refs: {e}");
                     }
                 }

     }
     /// Announce our own refs for the given repo.
     fn announce_own_refs(&mut self, rid: RepoId, doc: Doc) -> Result<Vec<RefsAt>, Error> {
         let (refs, timestamp) = self.announce_refs(rid, doc, [self.node_id()])?;
     fn announce_own_refs(
         &mut self,
         rid: RepoId,
         doc: Doc,
         namespaces: impl IntoIterator<Item = NodeId>,
     ) -> Result<(Vec<RefsAt>, Timestamp), Error> {
         let (refs, timestamp) = self.announce_refs(rid, doc, namespaces, true)?;
         // Update refs database with our signed refs branches.
         // This isn't strictly necessary for now, as we only use the database for fetches, and
         // we don't fetch our own refs that are announced, but it's for good measure.
         if let &[r] = refs.as_slice() {
         for r in refs.iter() {
             self.emitter.emit(Event::LocalRefsAnnounced {
                 rid,
                 refs: r,
                 refs: *r,
                 timestamp,
             });
             if let Err(e) = self.database_mut().refs_mut().set(

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

         rid: RepoId,
         doc: Doc,
         remotes: impl IntoIterator<Item = NodeId>,
         own: bool,
     ) -> Result<(Vec<RefsAt>, Timestamp), Error> {
         let (ann, refs) = self.refs_announcement_for(rid, remotes)?;
         let timestamp = ann.timestamp();

         // Update our sync status for our own refs. This is useful for determining if refs were
         // updated while the node was stopped.
         if let Some(refs) = refs.iter().find(|r| r.remote == ann.node) {
         for r in refs.iter().filter(|r| own || r.remote == ann.node) {
             info!(
                 target: "service",
                 "Announcing own refs for {rid} to peers ({}) (t={timestamp})..",
                 refs.at
                 "Announcing refs {rid}/{r} to peers (t={timestamp})..",
             );
             // Update our local node's sync status to mark the refs as announced.
             if let Err(e) = self
                 .db
                 .seeds_mut()
                 .synced(&rid, &ann.node, refs.at, timestamp)
             {
             if let Err(e) = self.db.seeds_mut().synced(&rid, &ann.node, r.at, timestamp) {
                 error!(target: "service", "Error updating sync status for local node: {e}");
             } else {
                 debug!(target: "service", "Saved local sync status for {rid}..");

         Ok(())
     }
     fn seeds(&self, rid: &RepoId) -> Result<Seeds, Error> {
     fn seeds(&self, rid: &RepoId, namespaces: HashSet<PublicKey>) -> Result<Seeds, Error> {
         let mut seeds = Seeds::new(self.rng.clone());
         // 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.
         // First, build a list of peers that have synced refs for `namespaces`, if any.
         // This step is skipped:
         //  1. For the repository (and thus all `namespaces`), if it not exist in storage.
         //  2. For each `namespace` in `namespaces`, which does not exist in storage.
         if let Ok(repo) = self.storage.repository(*rid) {
             if let Ok(local) = RefsAt::new(&repo, self.node_id()) {
             for namespace in namespaces.iter() {
                 let Ok(local) = RefsAt::new(&repo, *namespace) else {
                     continue;
                 };
                 for seed in self.db.seeds().seeds_for(rid)? {
                     let seed = seed?;
                     let state = self.sessions.get(&seed.nid).map(|s| s.state.clone());

         // These peers have announced that they seed the repository via an inventory
         // announcement, but we haven't received any ref announcements from them.
         for nid in self.db.routing().get(rid)? {
             if nid == self.node_id() {
             if namespaces.contains(&nid) {
                 continue;
             }
             if seeds.contains(&nid) {