eceb6a0 node: Report sync status for given public key — heartwood

modified crates/radicle-cli/src/commands/sync.rs

@@ -345,7 +345,7 @@ fn sync_status(

     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(rid, Some(*profile.did()))?.into();
     let local_nid = node.nid()?;
     let aliases = profile.aliases();

@@ -522,7 +522,7 @@ pub fn fetch(

         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(rid, None)?;
             let (connected, disconnected) = seeds.partition();
             let candidates = connected
                 .into_iter()

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

@@ -149,7 +149,7 @@ where

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

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

@@ -1704,7 +1704,7 @@ fn test_clone_without_seeds() {

     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(rid, None).unwrap();
     let connected = seeds.connected().collect::<Vec<_>>();
     assert!(connected.is_empty());

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

@@ -172,8 +172,8 @@ where

             CommandResult::Okay(addrs).to_writer(writer)?;
         }
         Command::Seeds { rid } => {
             let seeds = handle.seeds(rid)?;
         Command::Seeds { rid, ns } => {
             let seeds = handle.seeds(rid, ns)?;
             CommandResult::Okay(seeds).to_writer(writer)?;
         }

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

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

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

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

@@ -55,7 +55,7 @@ impl radicle::node::Handle for Handle {

         unimplemented!();
     }
     fn seeds(&mut self, _id: RepoId) -> Result<Seeds, Self::Error> {
     fn seeds(&mut self, _id: RepoId, _ns: Option<PublicKey>) -> Result<Seeds, Self::Error> {
         unimplemented!();
     }

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

@@ -200,7 +200,7 @@ impl<G: Signer<Signature> + cyphernet::Ecdh> NodeHandle<G> {

         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(*rid, None).unwrap();
             if seeds.iter().any(|s| s.nid == *nid && s.is_synced()) {
                 break;
             }

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

@@ -183,7 +183,7 @@ fn test_replication() {

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

@@ -549,7 +549,7 @@ fn test_clone() {

     transport::local::register(alice.storage.clone());
     let _ = alice.handle.seed(acme, Scope::All).unwrap();
     let seeds = alice.handle.seeds(acme).unwrap();
     let seeds = alice.handle.seeds(acme, None).unwrap();
     assert!(seeds.is_connected(&bob.id));
     let result = alice.handle.fetch(acme, bob.id, DEFAULT_TIMEOUT).unwrap();

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

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

     /// 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>),
     Seeds(RepoId, Option<PublicKey>, chan::Sender<Seeds>),
     /// Fetch the given repository from the network.
     Fetch(RepoId, NodeId, time::Duration, chan::Sender<FetchResult>),
     /// Seed the given repository.

@@ -278,7 +278,7 @@ impl fmt::Debug for Command {

             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})"),

@@ -880,20 +880,22 @@ where

             Command::ListenAddrs(resp) => {
                 resp.send(self.listening.clone()).ok();
             }
             Command::Seeds(rid, resp) => match self.seeds(&rid) {
                 Ok(seeds) => {
                     let (connected, disconnected) = seeds.partition();
                     debug!(
                         target: "service",
                         "Found {} connected seed(s) and {} disconnected seed(s) for {}",
                         connected.len(), disconnected.len(),  rid
                     );
                     resp.send(seeds).ok();
                 }
                 Err(e) => {
                     error!(target: "service", "Error getting seeds for {rid}: {e}");
             Command::Seeds(rid, ns, resp) => {
                 match self.seeds(&rid, ns.unwrap_or_else(|| self.node_id())) {
                     Ok(seeds) => {
                         let (connected, disconnected) = seeds.partition();
                         debug!(
                             target: "service",
                             "Found {} connected seed(s) and {} disconnected seed(s) for {}",
                             connected.len(), disconnected.len(),  rid
                         );
                         resp.send(seeds).ok();
                     }
                     Err(e) => {
                         error!(target: "service", "Error getting seeds for {rid}: {e}");
                     }
                 }
             },
             }
             Command::Fetch(rid, seed, timeout, resp) => {
                 self.fetch(rid, seed, timeout, Some(resp));
             }

@@ -2290,14 +2292,14 @@ where

         Ok(())
     }
     fn seeds(&self, rid: &RepoId) -> Result<Seeds, Error> {
     fn seeds(&self, rid: &RepoId, ns: 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.
         // First build a list from peers that have synced refs for `ns`, if any.
         // This step is skipped if we don't have the repository yet, or don't have
         // our own refs.
         // refs for `ns`.
         if let Ok(repo) = self.storage.repository(*rid) {
             if let Ok(local) = RefsAt::new(&repo, self.node_id()) {
             if let Ok(local) = RefsAt::new(&repo, ns) {
                 for seed in self.db.seeds().seeds_for(rid)? {
                     let seed = seed?;
                     let state = self.sessions.get(&seed.nid).map(|s| s.state.clone());

@@ -2320,7 +2322,7 @@ where

         // 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 nid == ns {
                 continue;
             }
             if seeds.contains(&nid) {

@@ -2538,7 +2540,7 @@ where

             if self.storage.contains(&rid)? {
                 continue;
             }
             match self.seeds(&rid) {
             match self.seeds(&rid, self.node_id()) {
                 Ok(seeds) => {
                     if let Some(connected) = NonEmpty::from_vec(seeds.connected().collect()) {
                         for seed in connected {

modified crates/radicle/src/node.rs

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

         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

@@ -668,7 +668,25 @@ pub enum Command {

     /// Lookup seeds for the given repository in the routing table.
     #[serde(rename_all = "camelCase")]
     Seeds { rid: RepoId },
     Seeds {
         rid: RepoId,
         /// The namespace for which sync status should be reported.
         ///
         /// 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 report sync with
         /// "its 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>,
     },
     /// Get the current peer sessions.
     Sessions,

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

     /// Disconnect from a peer.
     fn disconnect(&mut self, node: NodeId) -> Result<(), Self::Error>;
     /// Lookup the seeds of a given repository in the routing table.
     fn seeds(&mut self, id: RepoId) -> Result<Seeds, Self::Error>;
     fn seeds(&mut self, id: RepoId, ns: Option<PublicKey>) -> Result<Seeds, Self::Error>;
     /// Fetch a repository from the network.
     fn fetch(
         &mut self,

@@ -1353,9 +1371,9 @@ impl Handle for Node {

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