From 1f837c65723eb6afbcdb6f0be3ac882e6586d27a Mon Sep 17 00:00:00 2001 From: Fintan Halpenny Date: Wed, 22 Feb 2023 13:20:59 +0000 Subject: [PATCH] node: implement tracking and routing The routing function for Node was not implemented and there were no functions for getting any tracking information. This implements routing, tracked_repos, and tracked_nodes. To accomplish this it was necessary to add associated types to the Handler trait so that implementations could differ in the types they use for returning this data, i.e. the control socket needs to use a channel while the Node needs to return serializable data. Currently, the Node uses a `Vec` but it would be more beneficial to return an Iterator directly to ensure we are not always hitting the heap. To allow for the tracking types to be used across radicle-node (the crate) and in radicle::node (the module), it was necessary to move the types to the common place, i.e. radicle::node. At the same time, new structs were added to make it easier to refer to tracked repos vs tracked nodes. Signed-off-by: Fintan Halpenny X-Clacks-Overhead: GNU Terry Pratchett --- radicle-node/src/control.rs | 22 ++- radicle-node/src/runtime/handle.rs | 41 +++++- radicle-node/src/service.rs | 20 ++- radicle-node/src/service/tracking.rs | 63 +------- radicle-node/src/service/tracking/store.rs | 83 ++--------- radicle-node/src/test/handle.rs | 33 ++++- radicle/src/node.rs | 48 +++++- radicle/src/node/tracking.rs | 164 +++++++++++++++++++++ 8 files changed, 332 insertions(+), 142 deletions(-) create mode 100644 radicle/src/node/tracking.rs diff --git a/radicle-node/src/control.rs b/radicle-node/src/control.rs index f778f2da..2fb2542c 100644 --- a/radicle-node/src/control.rs +++ b/radicle-node/src/control.rs @@ -126,6 +126,15 @@ fn command>( } } } + 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>( } } } + 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>( } 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)), diff --git a/radicle-node/src/runtime/handle.rs b/radicle-node/src/runtime/handle.rs index fa223ca6..13822f02 100644 --- a/radicle-node/src/runtime/handle.rs +++ b/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 radicle::node::Handle for Handle { type Sessions = Sessions; type Error = Error; + type Routing = chan::Receiver<(Id, NodeId)>; + type TrackedRepos = chan::Receiver; + type TrackedNodes = chan::Receiver; + fn is_running(&self) -> bool { true } @@ -131,6 +136,40 @@ impl radicle::node::Handle for Handle { receiver.recv().map_err(Error::from) } + fn tracked_repos(&self) -> Result { + let (sender, receiver) = chan::unbounded(); + let query: Arc = 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 { + let (sender, receiver) = chan::unbounded(); + let query: Arc = 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) -> Result { let (sender, receiver) = chan::bounded(1); self.command(service::Command::TrackNode(id, alias, sender))?; @@ -169,7 +208,7 @@ impl radicle::node::Handle for Handle { receiver.recv().map_err(Error::from) } - fn routing(&self) -> Result, Error> { + fn routing(&self) -> Result { let (sender, receiver) = chan::unbounded(); let query: Arc = Arc::new(move |state| { for (id, node) in state.routing().entries()? { diff --git a/radicle-node/src/service.rs b/radicle-node/src/service.rs index 45a30fde..43e65a14 100644 --- a/radicle-node/src/service.rs +++ b/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, tracking::Error>; + /// Get the tracked nodes. + fn tracked_nodes(&self) -> Result, tracking::Error>; } impl ServiceState for Service @@ -1354,6 +1358,14 @@ where fn routing(&self) -> &dyn routing::Store { &self.routing } + + fn tracked_repos(&self) -> Result, tracking::Error> { + Ok(self.tracking.repo_entries()?.collect()) + } + + fn tracked_nodes(&self) -> Result, tracking::Error> { + Ok(self.tracking.node_entries()?.collect()) + } } /// Disconnect reason. diff --git a/radicle-node/src/service/tracking.rs b/radicle-node/src/service/tracking.rs index edc7ab33..363e577a 100644 --- a/radicle-node/src/service/tracking.rs +++ b/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 { - 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 { - match s { - "trusted" => Ok(Self::Trusted), - "delegates-only" => Ok(Self::DelegatesOnly), - "all" => Ok(Self::All), - _ => Err(()), - } - } -} - /// Tracking configuration. #[derive(Debug)] pub struct Config { diff --git a/radicle-node/src/service/tracking/store.rs b/radicle-node/src/service/tracking/store.rs index 46997f1c..5ac028d5 100644 --- a/radicle-node/src/service/tracking/store.rs +++ b/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(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 { - 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(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 { - 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>, Error> { + pub fn node_entries(&self) -> Result>, 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"); - 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>, Error> { + pub fn repo_entries(&self) -> Result>, 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"); - 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] diff --git a/radicle-node/src/test/handle.rs b/radicle-node/src/test/handle.rs index 1d97a692..ab0d7b29 100644 --- a/radicle-node/src/test/handle.rs +++ b/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; + type TrackedNodes = Vec; fn is_running(&self) -> bool { true @@ -37,6 +40,32 @@ impl radicle::node::Handle for Handle { Ok(FetchResult::from(Ok::, Self::Error>(vec![]))) } + fn tracked_repos(&self) -> Result { + 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 { + 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 { Ok(self.tracking_repos.insert(id)) } @@ -67,7 +96,7 @@ impl radicle::node::Handle for Handle { unimplemented!() } - fn routing(&self) -> Result, Self::Error> { + fn routing(&self) -> Result { unimplemented!(); } diff --git a/radicle/src/node.rs b/radicle/src/node.rs index 9efcfb2f..495cab54 100644 --- a/radicle/src/node.rs +++ b/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; + type TrackedRepos: IntoIterator; + type TrackedNodes: IntoIterator; + /// 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; /// Untrack the given node. fn untrack_node(&mut self, id: NodeId) -> Result; + /// Get the tracking information for all tracked repos in storage. + fn tracked_repos(&self) -> Result; + /// Get the tracking information for all tracked nodes in storage. + fn tracked_nodes(&self) -> Result; /// 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, Self::Error>; + fn routing(&self) -> Result; /// Query the peer session state. fn sessions(&self) -> Result; /// 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; + type TrackedNodes = Vec; + 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, 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, 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) -> Result { 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, Error> { - todo!(); + fn routing(&self) -> Result { + 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 { diff --git a/radicle/src/node/tracking.rs b/radicle/src/node/tracking.rs new file mode 100644 index 00000000..46262217 --- /dev/null +++ b/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 { + match s { + "track" => Ok(Self::Track), + "block" => Ok(Self::Block), + _ => Err(s.to_owned()), + } + } +} + +#[cfg(feature = "sql")] +impl sqlite::BindableWithIndex for Policy { + fn bind( + 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 { + 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 { + 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( + 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 { + 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, + }), + } + } +}