ec8bb59 REVIEW — 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, [*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, [*profile.did()])?;
             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, [*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

@@ -202,9 +202,13 @@ 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: 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(ns), sender))?;
         receiver.recv().map_err(Error::from)
     }

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

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

         unimplemented!();
     }
     fn seeds(&mut self, _id: RepoId) -> Result<Seeds, Self::Error> {
     fn seeds(
         &mut self,
         _id: RepoId,
         _ns: impl IntoIterator<Item = 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

@@ -252,8 +252,9 @@ pub enum Command {

     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.

@@ -278,7 +279,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 +881,30 @@ 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) => {
                 // For backwards compatibility default to own identity.
                 // See documentation of [`radicle::node::Command::Seeds`].
                 let ns = if ns.is_empty() {
                     [self.node_id()].into()
                 } else {
                     ns
                 };
                 match self.seeds(&rid, ns) {
                     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));
             }

@@ -2291,14 +2302,17 @@ where

         Ok(())
     }
     fn seeds(&self, rid: &RepoId) -> Result<Seeds, Error> {
     fn seeds(&self, rid: &RepoId, ns: 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 from peers that have synced refs for `ns`, if any.
         // This step is skipped if we don't have the repository yet.
         if let Ok(repo) = self.storage.repository(*rid) {
             if let Ok(local) = RefsAt::new(&repo, self.node_id()) {
             for namespace in ns.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());

@@ -2321,7 +2335,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 ns.contains(&nid) {
                 continue;
             }
             if seeds.contains(&nid) {

@@ -2539,7 +2553,7 @@ where

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

modified crates/radicle/src/node.rs

@@ -669,7 +669,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 namespaces for which sync status should be reported.
         ///
         /// For backwards compatibility, this is optional and defaults to the
         /// empty set. However, the empty set is interpreted by the node just
         /// like the singleton set containing the Node ID of the node
         /// itself, i.e., the node receives this command. 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 these identities are different.
         #[cfg_attr(
             feature = "schemars",
             schemars(with = "HashSet<crate::schemars_ext::crypto::PublicKey>")
         )]
         #[serde(default, skip_serializing_if = "HashSet::is_empty")]
         ns: HashSet<PublicKey>,
     },
     /// Get the current peer sessions.
     Sessions,

@@ -1115,7 +1133,11 @@ 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,
         rid: RepoId,
         ns: impl IntoIterator<Item = PublicKey>,
     ) -> Result<Seeds, Self::Error>;
     /// Fetch a repository from the network.
     fn fetch(
         &mut self,

@@ -1362,9 +1384,19 @@ impl Handle for Node {

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