1f837c6 node: implement tracking and routing — heartwood

Radicle Heartwood Protocol & Stack

node: implement tracking and routing

Fintan Halpenny committed 3 years ago

commit 1f837c65723eb6afbcdb6f0be3ac882e6586d27a
parent 900796956cef71b858dfcc2513a29cf3b9a2cf65

8 files changed +332 -142

modified radicle-node/src/control.rs

@@ -126,6 +126,15 @@ fn command<H: Handle<Error = runtime::HandleError>>(

                 }
             }
         }
         CommandName::TrackedRepos => match handle.tracked_repos() {
             Ok(ts) => {
                 for t in ts.into_iter() {
                     serde_json::to_writer(&mut writer, &t)?;
                     writeln!(writer)?;
                 }
             }
             Err(e) => return Err(CommandError::Runtime(e)),
         },
         CommandName::TrackNode => {
             let (node, alias) = match cmd.args.as_slice() {
                 [node] => (node.as_str(), None),

@@ -157,6 +166,15 @@ fn command<H: Handle<Error = runtime::HandleError>>(

                 }
             }
         }
         CommandName::TrackedNodes => match handle.tracked_nodes() {
             Ok(ts) => {
                 for t in ts.into_iter() {
                     serde_json::to_writer(&mut writer, &t)?;
                     writeln!(writer)?;
                 }
             }
             Err(e) => return Err(CommandError::Runtime(e)),
         },
         CommandName::AnnounceRefs => {
             let rid: Id = parse::arg(cmd)?;

@@ -184,8 +202,8 @@ fn command<H: Handle<Error = runtime::HandleError>>(

         }
         CommandName::Routing => match handle.routing() {
             Ok(c) => {
                 for (id, seed) in c.iter() {
                     writeln!(writer, "{id} {seed}")?;
                 for route in c.into_iter() {
                     serde_json::to_writer(&mut writer, &route)?;
                 }
             }
             Err(e) => return Err(CommandError::Runtime(e)),

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

@@ -14,6 +14,7 @@ use crate::identity::Id;

 use crate::node::{Command, FetchResult};
 use crate::profile::Home;
 use crate::service;
 use crate::service::tracking;
 use crate::service::{CommandError, QueryState};
 use crate::service::{NodeId, Sessions};
 use crate::wire;

@@ -109,6 +110,10 @@ impl<G: Signer + Ecdh + 'static> radicle::node::Handle for Handle<G> {

     type Sessions = Sessions;
     type Error = Error;
     type Routing = chan::Receiver<(Id, NodeId)>;
     type TrackedRepos = chan::Receiver<tracking::Repo>;
     type TrackedNodes = chan::Receiver<tracking::Node>;
     fn is_running(&self) -> bool {
         true
     }

@@ -131,6 +136,40 @@ impl<G: Signer + Ecdh + 'static> radicle::node::Handle for Handle<G> {

         receiver.recv().map_err(Error::from)
     }
     fn tracked_repos(&self) -> Result<Self::TrackedRepos, Self::Error> {
         let (sender, receiver) = chan::unbounded();
         let query: Arc<QueryState> = Arc::new(move |state| {
             for t in state.tracked_repos()? {
                 if sender.send(t).is_err() {
                     break;
                 }
             }
             Ok(())
         });
         let (err_sender, err_receiver) = chan::bounded(1);
         self.command(service::Command::QueryState(query, err_sender))?;
         err_receiver.recv()??;
         Ok(receiver)
     }
     fn tracked_nodes(&self) -> Result<Self::TrackedNodes, Self::Error> {
         let (sender, receiver) = chan::unbounded();
         let query: Arc<QueryState> = Arc::new(move |state| {
             for t in state.tracked_nodes()? {
                 if sender.send(t).is_err() {
                     break;
                 }
             }
             Ok(())
         });
         let (err_sender, err_receiver) = chan::bounded(1);
         self.command(service::Command::QueryState(query, err_sender))?;
         err_receiver.recv()??;
         Ok(receiver)
     }
     fn track_node(&mut self, id: NodeId, alias: Option<String>) -> Result<bool, Error> {
         let (sender, receiver) = chan::bounded(1);
         self.command(service::Command::TrackNode(id, alias, sender))?;

@@ -169,7 +208,7 @@ impl<G: Signer + Ecdh + 'static> radicle::node::Handle for Handle<G> {

         receiver.recv().map_err(Error::from)
     }
     fn routing(&self) -> Result<chan::Receiver<(Id, NodeId)>, Error> {
     fn routing(&self) -> Result<Self::Routing, Error> {
         let (sender, receiver) = chan::unbounded();
         let query: Arc<QueryState> = Arc::new(move |state| {
             for (id, node) in state.routing().entries()? {

modified radicle-node/src/service.rs

@@ -151,6 +151,8 @@ pub enum CommandError {

     Storage(#[from] storage::Error),
     #[error(transparent)]
     Routing(#[from] routing::Error),
     #[error(transparent)]
     Tracking(#[from] tracking::Error),
 }
 #[derive(Debug)]

@@ -285,8 +287,7 @@ where

         self.filter = Filter::new(
             self.tracking
                 .repo_entries()?
                 .filter(|(_, _, policy)| *policy == tracking::Policy::Track)
                 .map(|(e, _, _)| e),
                 .filter_map(|t| (t.policy == tracking::Policy::Track).then_some(t.id)),
         );
         Ok(updated)
     }

@@ -376,8 +377,7 @@ where

         self.filter = Filter::new(
             self.tracking
                 .repo_entries()?
                 .filter(|(_, _, policy)| *policy == tracking::Policy::Track)
                 .map(|(e, _, _)| e),
                 .filter_map(|t| (t.policy == tracking::Policy::Track).then_some(t.id)),
         );
         Ok(())

@@ -1319,6 +1319,10 @@ pub trait ServiceState {

     fn config(&self) -> &Config;
     /// Get reference to routing table.
     fn routing(&self) -> &dyn routing::Store;
     /// Get the tracked repos.
     fn tracked_repos(&self) -> Result<Vec<tracking::Repo>, tracking::Error>;
     /// Get the tracked nodes.
     fn tracked_nodes(&self) -> Result<Vec<tracking::Node>, tracking::Error>;
 }
 impl<R, A, S, G> ServiceState for Service<R, A, S, G>

@@ -1354,6 +1358,14 @@ where

     fn routing(&self) -> &dyn routing::Store {
         &self.routing
     }
     fn tracked_repos(&self) -> Result<Vec<tracking::Repo>, tracking::Error> {
         Ok(self.tracking.repo_entries()?.collect())
     }
     fn tracked_nodes(&self) -> Result<Vec<tracking::Node>, tracking::Error> {
         Ok(self.tracking.node_entries()?.collect())
     }
 }
 /// Disconnect reason.

modified radicle-node/src/service/tracking.rs

@@ -1,72 +1,15 @@

 mod store;
 use std::str::FromStr;
 use std::{fmt, ops};
 use std::ops;
 use crate::prelude::Id;
 use crate::service::NodeId;
 pub use crate::node::tracking::{Alias, Node, Policy, Repo, Scope};
 pub use store::Config as Store;
 pub use store::Error;
 /// Node alias.
 pub type Alias = String;
 /// Tracking policy.
 #[derive(Default, Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
 pub enum Policy {
     /// The resource is tracked.
     Track,
     /// The resource is blocked.
     #[default]
     Block,
 }
 impl fmt::Display for Policy {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
             Self::Track => write!(f, "track"),
             Self::Block => write!(f, "block"),
         }
     }
 }
 impl FromStr for Policy {
     type Err = String;
     fn from_str(s: &str) -> Result<Self, Self::Err> {
         match s {
             "track" => Ok(Self::Track),
             "block" => Ok(Self::Block),
             _ => Err(s.to_owned()),
         }
     }
 }
 /// Tracking scope of a repository tracking policy.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
 pub enum Scope {
     /// Track remotes of nodes that are already tracked.
     Trusted,
     /// Track remotes of repository delegates.
     DelegatesOnly,
     /// Track all remotes.
     All,
 }
 impl FromStr for Scope {
     type Err = ();
     fn from_str(s: &str) -> Result<Self, Self::Err> {
         match s {
             "trusted" => Ok(Self::Trusted),
             "delegates-only" => Ok(Self::DelegatesOnly),
             "all" => Ok(Self::All),
             _ => Err(()),
         }
     }
 }
 /// Tracking configuration.
 #[derive(Debug)]
 pub struct Config {

modified radicle-node/src/service/tracking/store.rs

@@ -1,6 +1,5 @@

 #![allow(clippy::type_complexity)]
 use std::path::Path;
 use std::str::FromStr;
 use std::{fmt, io};
 use sqlite as sql;

@@ -9,7 +8,7 @@ use thiserror::Error;

 use crate::prelude::Id;
 use crate::service::NodeId;
 use super::{Alias, Policy, Scope};
 use super::{Alias, Node, Policy, Repo, Scope};
 #[derive(Error, Debug)]
 pub enum Error {

@@ -21,63 +20,6 @@ pub enum Error {

     Internal(#[from] sql::Error),
 }
 impl sqlite::BindableWithIndex for Scope {
     fn bind<I: sql::ParameterIndex>(self, stmt: &mut sql::Statement<'_>, i: I) -> sql::Result<()> {
         let s = match self {
             Self::Trusted => "trusted",
             Self::DelegatesOnly => "delegates-only",
             Self::All => "all",
         };
         s.bind(stmt, i)
     }
 }
 impl TryFrom<&sql::Value> for Scope {
     type Error = sql::Error;
     fn try_from(value: &sql::Value) -> Result<Self, Self::Error> {
         let message = Some("invalid remote scope".to_owned());
         match value {
             sql::Value::String(scope) => Scope::from_str(scope).map_err(|_| sql::Error {
                 code: None,
                 message,
             }),
             _ => Err(sql::Error {
                 code: None,
                 message,
             }),
         }
     }
 }
 impl sqlite::BindableWithIndex for Policy {
     fn bind<I: sql::ParameterIndex>(self, stmt: &mut sql::Statement<'_>, i: I) -> sql::Result<()> {
         match self {
             Self::Track => "track",
             Self::Block => "block",
         }
         .bind(stmt, i)
     }
 }
 impl TryFrom<&sql::Value> for Policy {
     type Error = sql::Error;
     fn try_from(value: &sql::Value) -> Result<Self, Self::Error> {
         let message = Some("sql: invalid policy".to_owned());
         match value {
             sql::Value::String(s) if s == "track" => Ok(Policy::Track),
             sql::Value::String(s) if s == "block" => Ok(Policy::Block),
             _ => Err(sql::Error {
                 code: None,
                 message,
             }),
         }
     }
 }
 /// Tracking configuration.
 pub struct Config {
     db: sql::Connection,

@@ -248,7 +190,7 @@ impl Config {

     }
     /// Get node tracking entries.
     pub fn node_entries(&self) -> Result<Box<dyn Iterator<Item = (NodeId, Alias, Policy)>>, Error> {
     pub fn node_entries(&self) -> Result<Box<dyn Iterator<Item = Node>>, Error> {
         let mut stmt = self
             .db
             .prepare("SELECT id, alias, policy FROM `node-policies`")?

@@ -257,16 +199,17 @@ impl Config {

         while let Some(Ok(row)) = stmt.next() {
             let id = row.read("id");
             let alias = row.read::<&str, _>("alias");
             let alias = row.read::<&str, _>("alias").to_owned();
             let policy = row.read::<Policy, _>("policy");
             entries.push((id, alias.to_owned(), policy));
             entries.push(Node { id, alias, policy });
         }
         Ok(Box::new(entries.into_iter()))
     }
     // TODO: see if sql can return iterator directly
     /// Get repository tracking entries.
     pub fn repo_entries(&self) -> Result<Box<dyn Iterator<Item = (Id, Scope, Policy)>>, Error> {
     pub fn repo_entries(&self) -> Result<Box<dyn Iterator<Item = Repo>>, Error> {
         let mut stmt = self
             .db
             .prepare("SELECT id, scope, policy FROM `repo-policies`")?

@@ -278,7 +221,7 @@ impl Config {

             let scope = row.read("scope");
             let policy = row.read::<Policy, _>("policy");
             entries.push((id, scope, policy));
             entries.push(Repo { id, scope, policy });
         }
         Ok(Box::new(entries.into_iter()))
     }

@@ -324,9 +267,9 @@ mod test {

             assert!(db.track_node(id, None).unwrap());
         }
         let mut entries = db.node_entries().unwrap();
         assert_matches!(entries.next(), Some((id, _, _)) if id == ids[0]);
         assert_matches!(entries.next(), Some((id, _, _)) if id == ids[1]);
         assert_matches!(entries.next(), Some((id, _, _)) if id == ids[2]);
         assert_matches!(entries.next(), Some(Node { id, .. }) if id == ids[0]);
         assert_matches!(entries.next(), Some(Node { id, .. }) if id == ids[1]);
         assert_matches!(entries.next(), Some(Node { id, .. }) if id == ids[2]);
     }
     #[test]

@@ -338,9 +281,9 @@ mod test {

             assert!(db.track_repo(id, Scope::All).unwrap());
         }
         let mut entries = db.repo_entries().unwrap();
         assert_matches!(entries.next(), Some((id, _, _)) if id == ids[0]);
         assert_matches!(entries.next(), Some((id, _, _)) if id == ids[1]);
         assert_matches!(entries.next(), Some((id, _, _)) if id == ids[2]);
         assert_matches!(entries.next(), Some(Repo { id, .. }) if id == ids[0]);
         assert_matches!(entries.next(), Some(Repo { id, .. }) if id == ids[1]);
         assert_matches!(entries.next(), Some(Repo { id, .. }) if id == ids[2]);
     }
     #[test]

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

@@ -6,8 +6,8 @@ use crossbeam_channel as chan;

 use crate::identity::Id;
 use crate::node::{FetchResult, Seeds};
 use crate::runtime::HandleError;
 use crate::service;
 use crate::service::NodeId;
 use crate::service::{self, tracking};
 use crate::storage::RefUpdate;
 #[derive(Default, Clone)]

@@ -20,6 +20,9 @@ pub struct Handle {

 impl radicle::node::Handle for Handle {
     type Error = HandleError;
     type Sessions = service::Sessions;
     type Routing = Vec<(Id, NodeId)>;
     type TrackedRepos = Vec<tracking::Repo>;
     type TrackedNodes = Vec<tracking::Node>;
     fn is_running(&self) -> bool {
         true

@@ -37,6 +40,32 @@ impl radicle::node::Handle for Handle {

         Ok(FetchResult::from(Ok::<Vec<RefUpdate>, Self::Error>(vec![])))
     }
     fn tracked_repos(&self) -> Result<Self::TrackedRepos, Self::Error> {
         Ok(self
             .tracking_repos
             .iter()
             .copied()
             .map(|id| tracking::Repo {
                 id,
                 scope: tracking::Scope::All,
                 policy: tracking::Policy::Track,
             })
             .collect())
     }
     fn tracked_nodes(&self) -> Result<Self::TrackedNodes, Self::Error> {
         Ok(self
             .tracking_nodes
             .iter()
             .copied()
             .map(|id| tracking::Node {
                 id,
                 alias: "".to_string(),
                 policy: tracking::Policy::Track,
             })
             .collect())
     }
     fn track_repo(&mut self, id: Id) -> Result<bool, Self::Error> {
         Ok(self.tracking_repos.insert(id))
     }

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

         unimplemented!()
     }
     fn routing(&self) -> Result<chan::Receiver<(Id, service::NodeId)>, Self::Error> {
     fn routing(&self) -> Result<Self::Routing, Self::Error> {
         unimplemented!();
     }

modified radicle/src/node.rs

@@ -1,4 +1,5 @@

 mod features;
 pub mod tracking;
 use std::collections::BTreeSet;
 use std::io::{BufRead, BufReader};

@@ -124,10 +125,14 @@ pub enum CommandName {

     TrackRepo,
     /// Untrack the given repository.
     UntrackRepo,
     /// Get the tracked repositories.
     TrackedRepos,
     /// Track the given node.
     TrackNode,
     /// Untrack the given node.
     UntrackNode,
     /// Get the tracked nodes.
     TrackedNodes,
     /// Get the node's inventory.
     Inventory,
     /// Get the node's routing table.

@@ -374,6 +379,10 @@ pub trait Handle {

     /// The error returned by all methods.
     type Error: std::error::Error + Send + Sync + 'static;
     type Routing: IntoIterator<Item = (Id, NodeId)>;
     type TrackedRepos: IntoIterator<Item = tracking::Repo>;
     type TrackedNodes: IntoIterator<Item = tracking::Node>;
     /// Check if the node is running. to a peer.
     fn is_running(&self) -> bool;
     /// Connect to a peer.

@@ -391,6 +400,10 @@ pub trait Handle {

     fn untrack_repo(&mut self, id: Id) -> Result<bool, Self::Error>;
     /// Untrack the given node.
     fn untrack_node(&mut self, id: NodeId) -> Result<bool, Self::Error>;
     /// Get the tracking information for all tracked repos in storage.
     fn tracked_repos(&self) -> Result<Self::TrackedRepos, Self::Error>;
     /// Get the tracking information for all tracked nodes in storage.
     fn tracked_nodes(&self) -> Result<Self::TrackedNodes, Self::Error>;
     /// Notify the service that a project has been updated, and announce local refs.
     fn announce_refs(&mut self, id: Id) -> Result<(), Self::Error>;
     /// Announce local inventory.

@@ -400,7 +413,7 @@ pub trait Handle {

     /// Ask the service to shutdown.
     fn shutdown(self) -> Result<(), Self::Error>;
     /// Query the routing table entries.
     fn routing(&self) -> Result<chan::Receiver<(Id, NodeId)>, Self::Error>;
     fn routing(&self) -> Result<Self::Routing, Self::Error>;
     /// Query the peer session state.
     fn sessions(&self) -> Result<Self::Sessions, Self::Error>;
     /// Query the inventory.

@@ -446,10 +459,16 @@ impl Node {

     }
 }
 // TODO(finto): tracked_repos, tracked_nodes, and routing should all
 // attempt to return iterators instead of allocating vecs.
 impl Handle for Node {
     type Sessions = ();
     type Error = Error;
     type TrackedRepos = Vec<tracking::Repo>;
     type TrackedNodes = Vec<tracking::Node>;
     type Routing = Vec<(Id, NodeId)>;
     fn is_running(&self) -> bool {
         let Ok(mut lines) = self.call::<&str, CommandResult>(CommandName::Status, []) else {
             return false;

@@ -486,6 +505,24 @@ impl Handle for Node {

         Ok(result)
     }
     fn tracked_repos(&self) -> Result<Vec<tracking::Repo>, Self::Error> {
         let mut repos = Vec::new();
         for result in self.call::<&str, _>(CommandName::TrackedRepos, [])? {
             let repo = result?;
             repos.push(repo);
         }
         Ok(repos)
     }
     fn tracked_nodes(&self) -> Result<Vec<tracking::Node>, Self::Error> {
         let mut repos = Vec::new();
         for result in self.call::<&str, _>(CommandName::TrackedNodes, [])? {
             let repo = result?;
             repos.push(repo);
         }
         Ok(repos)
     }
     fn track_node(&mut self, id: NodeId, alias: Option<String>) -> Result<bool, Error> {
         let id = id.to_human();
         let args = if let Some(alias) = alias.as_deref() {

@@ -552,8 +589,13 @@ impl Handle for Node {

         response.into()
     }
     fn routing(&self) -> Result<chan::Receiver<(Id, NodeId)>, Error> {
         todo!();
     fn routing(&self) -> Result<Self::Routing, Error> {
         let mut routes = Vec::new();
         for result in self.call::<&str, _>(CommandName::Routing, [])? {
             let route = result?;
             routes.push(route);
         }
         Ok(routes)
     }
     fn sessions(&self) -> Result<Self::Sessions, Error> {

added radicle/src/node/tracking.rs

@@ -0,0 +1,164 @@

 use std::fmt;
 use std::str::FromStr;
 use serde::{Deserialize, Serialize};
 use thiserror::Error;
 use crate::prelude::Id;
 use super::NodeId;
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub struct Repo {
     pub id: Id,
     pub scope: Scope,
     pub policy: Policy,
 }
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub struct Node {
     pub id: NodeId,
     pub alias: Alias,
     pub policy: Policy,
 }
 /// Node alias.
 pub type Alias = String;
 /// Tracking policy.
 #[derive(Default, Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
 pub enum Policy {
     /// The resource is tracked.
     Track,
     /// The resource is blocked.
     #[default]
     Block,
 }
 impl fmt::Display for Policy {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
             Self::Track => write!(f, "track"),
             Self::Block => write!(f, "block"),
         }
     }
 }
 impl FromStr for Policy {
     type Err = String;
     fn from_str(s: &str) -> Result<Self, Self::Err> {
         match s {
             "track" => Ok(Self::Track),
             "block" => Ok(Self::Block),
             _ => Err(s.to_owned()),
         }
     }
 }
 #[cfg(feature = "sql")]
 impl sqlite::BindableWithIndex for Policy {
     fn bind<I: sqlite::ParameterIndex>(
         self,
         stmt: &mut sqlite::Statement<'_>,
         i: I,
     ) -> sqlite::Result<()> {
         match self {
             Self::Track => "track",
             Self::Block => "block",
         }
         .bind(stmt, i)
     }
 }
 #[cfg(feature = "sql")]
 impl TryFrom<&sqlite::Value> for Policy {
     type Error = sqlite::Error;
     fn try_from(value: &sqlite::Value) -> Result<Self, Self::Error> {
         let message = Some("sql: invalid policy".to_owned());
         match value {
             sqlite::Value::String(s) if s == "track" => Ok(Policy::Track),
             sqlite::Value::String(s) if s == "block" => Ok(Policy::Block),
             _ => Err(sqlite::Error {
                 code: None,
                 message,
             }),
         }
     }
 }
 /// Tracking scope of a repository tracking policy.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
 pub enum Scope {
     /// Track remotes of nodes that are already tracked.
     Trusted,
     /// Track remotes of repository delegates.
     DelegatesOnly,
     /// Track all remotes.
     All,
 }
 impl fmt::Display for Scope {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
             Scope::Trusted => f.write_str("trusted"),
             Scope::DelegatesOnly => f.write_str("delegates-only"),
             Scope::All => f.write_str("all"),
         }
     }
 }
 #[derive(Debug, Error)]
 #[error("invalid tracking scope: {0:?}")]
 pub struct ParseScopeError(String);
 impl FromStr for Scope {
     type Err = ParseScopeError;
     fn from_str(s: &str) -> Result<Self, Self::Err> {
         match s {
             "trusted" => Ok(Self::Trusted),
             "delegates-only" => Ok(Self::DelegatesOnly),
             "all" => Ok(Self::All),
             _ => Err(ParseScopeError(s.to_string())),
         }
     }
 }
 #[cfg(feature = "sql")]
 impl sqlite::BindableWithIndex for Scope {
     fn bind<I: sqlite::ParameterIndex>(
         self,
         stmt: &mut sqlite::Statement<'_>,
         i: I,
     ) -> sqlite::Result<()> {
         let s = match self {
             Self::Trusted => "trusted",
             Self::DelegatesOnly => "delegates-only",
             Self::All => "all",
         };
         s.bind(stmt, i)
     }
 }
 #[cfg(feature = "sql")]
 impl TryFrom<&sqlite::Value> for Scope {
     type Error = sqlite::Error;
     fn try_from(value: &sqlite::Value) -> Result<Self, Self::Error> {
         let message = Some("invalid remote scope".to_owned());
         match value {
             sqlite::Value::String(scope) => Scope::from_str(scope).map_err(|_| sqlite::Error {
                 code: None,
                 message,
             }),
             _ => Err(sqlite::Error {
                 code: None,
                 message,
             }),
         }
     }
 }