From 1007756238a18909c95d00bf7484101e87097683 Mon Sep 17 00:00:00 2001 From: Alexis Sellier Date: Tue, 6 Dec 2022 14:56:36 +0100 Subject: [PATCH] node: Add tracking configuration store Signed-off-by: Alexis Sellier --- radicle-crypto/src/lib.rs | 2 +- radicle-node/src/address/store.rs | 36 +- radicle-node/src/client.rs | 12 +- radicle-node/src/service.rs | 89 +++-- radicle-node/src/service/config.rs | 76 +--- radicle-node/src/service/filter.rs | 15 +- radicle-node/src/service/tracking.rs | 38 ++ radicle-node/src/service/tracking/schema.sql | 32 ++ radicle-node/src/service/tracking/store.rs | 397 +++++++++++++++++++ radicle-node/src/test/peer.rs | 12 +- radicle-node/src/tests.rs | 39 +- radicle/src/sql.rs | 4 +- 12 files changed, 606 insertions(+), 146 deletions(-) create mode 100644 radicle-node/src/service/tracking.rs create mode 100644 radicle-node/src/service/tracking/schema.sql create mode 100644 radicle-node/src/service/tracking/store.rs diff --git a/radicle-crypto/src/lib.rs b/radicle-crypto/src/lib.rs index 19ae9db6..58b17547 100644 --- a/radicle-crypto/src/lib.rs +++ b/radicle-crypto/src/lib.rs @@ -369,7 +369,7 @@ impl TryFrom<&sqlite::Value> for PublicKey { }), _ => Err(sqlite::Error { code: None, - message: None, + message: Some("sql: invalid type for public key".to_owned()), }), } } diff --git a/radicle-node/src/address/store.rs b/radicle-node/src/address/store.rs index 0fae0c8c..e275a96e 100644 --- a/radicle-node/src/address/store.rs +++ b/radicle-node/src/address/store.rs @@ -31,7 +31,7 @@ pub struct Book { impl fmt::Debug for Book { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - write!(f, "Cache(..)") + write!(f, "Book(..)") } } @@ -234,13 +234,13 @@ impl TryFrom<&sql::Value> for Address { fn try_from(value: &sql::Value) -> Result { match value { - sql::Value::String(s) => Address::from_str(s.as_str()).map_err(|_| sql::Error { + sql::Value::String(s) => Address::from_str(s.as_str()).map_err(|e| sql::Error { code: None, - message: None, + message: Some(e.to_string()), }), _ => Err(sql::Error { code: None, - message: None, + message: Some("sql: invalid type for address".to_owned()), }), } } @@ -256,20 +256,18 @@ impl TryFrom<&sql::Value> for Source { type Error = sql::Error; fn try_from(value: &sql::Value) -> Result { + let err = sql::Error { + code: None, + message: Some("sql: invalid source".to_owned()), + }; match value { sql::Value::String(s) => match s.as_str() { "dns" => Ok(Source::Dns), "peer" => Ok(Source::Peer), "imported" => Ok(Source::Imported), - _ => Err(sql::Error { - code: None, - message: None, - }), + _ => Err(err), }, - _ => Err(sql::Error { - code: None, - message: None, - }), + _ => Err(err), } } } @@ -288,21 +286,19 @@ impl TryFrom<&sql::Value> for AddressType { type Error = sql::Error; fn try_from(value: &sql::Value) -> Result { + let err = sql::Error { + code: None, + message: Some("sql: invalid address type".to_owned()), + }; match value { sql::Value::String(s) => match s.as_str() { "ipv4" => Ok(AddressType::Ipv4), "ipv6" => Ok(AddressType::Ipv6), "hostname" => Ok(AddressType::Hostname), "onion" => Ok(AddressType::Onion), - _ => Err(sql::Error { - code: None, - message: None, - }), + _ => Err(err), }, - _ => Err(sql::Error { - code: None, - message: None, - }), + _ => Err(err), } } } diff --git a/radicle-node/src/client.rs b/radicle-node/src/client.rs index 49abd3f0..5e344a46 100644 --- a/radicle-node/src/client.rs +++ b/radicle-node/src/client.rs @@ -8,7 +8,7 @@ use radicle::crypto::Signer; use crate::clock::RefClock; use crate::profile::Profile; -use crate::service::routing; +use crate::service::{routing, tracking}; use crate::wire::transcode::NoHandshake; use crate::wire::Wire; use crate::{address, service}; @@ -21,6 +21,8 @@ pub const NODE_DIR: &str = "node"; pub const ROUTING_DB_FILE: &str = "routing.db"; /// Filename of address database under [`NODE_DIR`]. pub const ADDRESS_DB_FILE: &str = "addresses.db"; +/// Filename of tracking table database under [`NODE_DIR`]. +pub const TRACKING_DB_FILE: &str = "tracking.db"; /// A client error. #[derive(Error, Debug)] @@ -31,6 +33,9 @@ pub enum Error { /// An address database error. #[error("address database error: {0}")] Addresses(#[from] address::Error), + /// A tracking database error. + #[error("tracking database error: {0}")] + Tracking(#[from] tracking::Error), /// An I/O error. #[error("i/o error: {0}")] Io(#[from] io::Error), @@ -111,6 +116,7 @@ impl Client { let node_dir = profile.home.join(NODE_DIR); let address_db = node_dir.join(ADDRESS_DB_FILE); let routing_db = node_dir.join(ROUTING_DB_FILE); + let tracking_db = node_dir.join(TRACKING_DB_FILE); log::info!("Opening address book {}..", address_db.display()); let addresses = address::Book::open(address_db)?; @@ -118,6 +124,9 @@ impl Client { log::info!("Opening routing table {}..", routing_db.display()); let routing = routing::Table::open(routing_db)?; + log::info!("Opening tracking policy table {}..", tracking_db.display()); + let tracking = tracking::Config::open(tracking_db)?; + log::info!("Initializing client ({:?})..", network); let service = service::Service::new( @@ -126,6 +135,7 @@ impl Client { routing, storage, addresses, + tracking, signer, rng, ); diff --git a/radicle-node/src/service.rs b/radicle-node/src/service.rs index 0c044b85..5c49311f 100644 --- a/radicle-node/src/service.rs +++ b/radicle-node/src/service.rs @@ -1,9 +1,11 @@ +#![allow(clippy::too_many_arguments)] pub mod config; pub mod filter; pub mod message; pub mod reactor; pub mod routing; pub mod session; +pub mod tracking; use std::collections::hash_map::Entry; use std::collections::{BTreeMap, HashMap, HashSet}; @@ -30,7 +32,6 @@ use crate::git; use crate::identity::{Doc, Id}; use crate::node; use crate::prelude::*; -use crate::service::config::ProjectTracking; use crate::service::message::{Address, Announcement, AnnouncementMessage, Ping}; use crate::service::message::{NodeAnnouncement, RefsAnnouncement}; use crate::storage; @@ -190,6 +191,8 @@ pub struct Service { routing: R, /// Node address manager. addresses: A, + /// Tracking policy configuration. + tracking: tracking::Config, /// State relating to gossip. gossip: Gossip, /// Peer sessions, currently or recently connected. @@ -204,6 +207,8 @@ pub struct Service { rng: Rng, /// Whether our local inventory no long represents what we have announced to the network. out_of_sync: bool, + /// Current tracked repository bloom filter. + filter: Filter, /// Last time the service was idle. last_idle: LocalTime, /// Last time the service synced. @@ -244,6 +249,7 @@ where routing: R, storage: S, addresses: A, + tracking: tracking::Config, signer: G, rng: Rng, ) -> Self { @@ -253,6 +259,7 @@ where config, storage, addresses, + tracking, signer, rng, clock, @@ -263,6 +270,7 @@ where reactor: Reactor::default(), sessions, out_of_sync: false, + filter: Filter::empty(), last_idle: LocalTime::default(), last_sync: LocalTime::default(), last_prune: LocalTime::default(), @@ -282,34 +290,30 @@ where } } - pub fn tracked(&self) -> Result, storage::Error> { - let tracked = match &self.config.project_tracking { - ProjectTracking::All { blocked } => self - .storage - .inventory()? - .into_iter() - .filter(|id| !blocked.contains(id)) - .collect(), - - ProjectTracking::Allowed(projs) => projs.iter().cloned().collect(), - }; - - Ok(tracked) - } - /// Track a project. /// Returns whether or not the tracking policy was updated. - pub fn track(&mut self, id: Id) -> bool { - self.out_of_sync = self.config.track(id); - self.out_of_sync + pub fn track(&mut self, id: &Id, scope: tracking::Scope) -> Result { + self.out_of_sync = self.tracking.track_repo(id, scope)?; + self.filter.insert(id); + + Ok(self.out_of_sync) } /// Untrack a project. /// Returns whether or not the tracking policy was updated. /// Note that when untracking, we don't announce anything to the network. This is because by /// simply not announcing it anymore, it will eventually be pruned by nodes. - pub fn untrack(&mut self, id: Id) -> bool { - self.config.untrack(id) + pub fn untrack(&mut self, id: &Id) -> Result { + // Nb. This is potentially slow if we have lots of projects. We should probably + // only re-compute the filter when we've untracked a certain amount of projects + // and the filter is really out of date. + self.filter = Filter::new(self.tracking.repo_entries()?.map(|(e, _)| e)); + self.tracking.untrack_repo(id) + } + + /// Check whether we are tracking a certain repository. + pub fn is_tracking(&self, id: &Id) -> Result { + self.tracking.is_repo_tracked(id) } /// Find the closest `n` peers by proximity in tracking graphs. @@ -340,6 +344,11 @@ where &mut self.storage } + /// Get the tracking policy. + pub fn tracking(&self) -> &tracking::Config { + &self.tracking + } + /// Get the local signer. pub fn signer(&self) -> &G { &self.signer @@ -425,7 +434,11 @@ where match cmd { Command::Connect(addr) => self.reactor.connect(addr), Command::Fetch(id, resp) => { - if !self.config.is_tracking(&id) { + if !self + .tracking + .is_repo_tracked(&id) + .expect("Service::command: error accessing tracking configuration") + { resp.send(FetchLookup::NotTracking).ok(); return; } @@ -479,10 +492,16 @@ where } } Command::Track(id, resp) => { - resp.send(self.track(id)).ok(); + let tracked = self + .track(&id, tracking::Scope::All) + .expect("Service::command: error tracking repository"); + resp.send(tracked).ok(); } Command::Untrack(id, resp) => { - resp.send(self.untrack(id)).ok(); + let untracked = self + .untrack(&id) + .expect("Service::command: error untracking repository"); + resp.send(untracked).ok(); } Command::AnnounceRefs(id) => { if let Err(err) = self.announce_refs(id) { @@ -531,6 +550,7 @@ where self.clock.timestamp(), &self.storage, &self.signer, + self.filter.clone(), &self.config, ), ); @@ -668,7 +688,11 @@ where AnnouncementMessage::Refs(message) => { // TODO: Buffer/throttle fetches. // TODO: Check that we're tracking this user as well. - if self.config.is_tracking(&message.id) { + if self + .tracking + .is_repo_tracked(&message.id) + .expect("Service::handle_announcement: error accessing tracking configuration") + { // Discard inventory messages we've already seen, otherwise update // out last seen time. if !peer.refs_announced(message.id, timestamp) { @@ -705,6 +729,11 @@ where if is_updated { return Ok(relay); } + } else { + log::debug!( + "Ignoring refs announcement from {announcer}: repository {} isn't tracked", + message.id + ); } } AnnouncementMessage::Node( @@ -795,6 +824,7 @@ where self.clock.timestamp(), &self.storage, &self.signer, + self.filter.clone(), &self.config, ), ); @@ -890,7 +920,11 @@ where let mut included = HashSet::new(); for proj_id in inventory { included.insert(proj_id); - if self.routing.insert(*proj_id, from, *timestamp)? && self.config.is_tracking(proj_id) + if self.routing.insert(*proj_id, from, *timestamp)? + && self + .tracking + .is_repo_tracked(proj_id) + .expect("Service::process_inventory: error accessing tracking configuration") { log::info!("Routing table updated for {} with seed {}", proj_id, from); } @@ -1264,6 +1298,7 @@ mod gossip { timestamp: Timestamp, storage: &S, signer: &G, + filter: Filter, config: &Config, ) -> Vec { let inventory = match storage.inventory() { @@ -1286,7 +1321,7 @@ mod gossip { .expect("external addresses are within the limit"), ), Message::inventory(gossip::inventory(timestamp, inventory), signer), - Message::subscribe(config.filter(), timestamp, Timestamp::MAX), + Message::subscribe(filter, timestamp, Timestamp::MAX), ]; if let Some(m) = gossip::node(timestamp, config) { msgs.push(Message::node(m, signer)); diff --git a/radicle-node/src/service/config.rs b/radicle-node/src/service/config.rs index 9c717f2e..6b62e31f 100644 --- a/radicle-node/src/service/config.rs +++ b/radicle-node/src/service/config.rs @@ -1,8 +1,5 @@ use super::nakamoto::LocalDuration; -use crate::collections::HashSet; -use crate::identity::{Id, PublicKey}; -use crate::service::filter::Filter; use crate::service::message::Address; /// Peer-to-peer network. @@ -13,35 +10,6 @@ pub enum Network { Test, } -/// Project tracking policy. -#[derive(Debug, Clone)] -pub enum ProjectTracking { - /// Track all projects we come across. - All { blocked: HashSet }, - /// Track a static list of projects. - Allowed(HashSet), -} - -impl Default for ProjectTracking { - fn default() -> Self { - Self::All { - blocked: HashSet::default(), - } - } -} - -/// Project remote tracking policy. -#[derive(Debug, Default, Clone)] -pub enum RemoteTracking { - /// Only track remotes of project delegates. - #[default] - DelegatesOnly, - /// Track all remotes. - All { blocked: HashSet }, - /// Track a specific list of users as well as the project delegates. - Allowed(HashSet), -} - /// Configuration parameters defining attributes of minima and maxima. #[derive(Debug, Clone)] pub struct Limits { @@ -70,14 +38,11 @@ pub struct Config { pub external_addresses: Vec
, /// Peer-to-peer network. pub network: Network, - /// Project tracking policy. - pub project_tracking: ProjectTracking, - /// Project remote tracking policy. - pub remote_tracking: RemoteTracking, /// Whether or not our node should relay inventories. pub relay: bool, /// List of addresses to listen on for protocol connections. pub listen: Vec
, + /// Configured service limits. pub limits: Limits, } @@ -87,8 +52,6 @@ impl Default for Config { connect: Vec::default(), external_addresses: Vec::default(), network: Network::default(), - project_tracking: ProjectTracking::default(), - remote_tracking: RemoteTracking::default(), relay: true, listen: vec![], limits: Limits::default(), @@ -97,40 +60,17 @@ impl Default for Config { } impl Config { + pub fn new(network: Network) -> Self { + Self { + network, + ..Self::default() + } + } + pub fn is_persistent(&self, addr: &Address) -> bool { self.connect.contains(addr) } - pub fn is_tracking(&self, id: &Id) -> bool { - match &self.project_tracking { - ProjectTracking::All { blocked } => !blocked.contains(id), - ProjectTracking::Allowed(ids) => ids.contains(id), - } - } - - /// Track a project. Returns whether the policy was updated. - pub fn track(&mut self, id: Id) -> bool { - match &mut self.project_tracking { - ProjectTracking::All { .. } => false, - ProjectTracking::Allowed(ids) => ids.insert(id), - } - } - - /// Untrack a project. Returns whether the policy was updated. - pub fn untrack(&mut self, id: Id) -> bool { - match &mut self.project_tracking { - ProjectTracking::All { blocked } => blocked.insert(id), - ProjectTracking::Allowed(ids) => ids.remove(&id), - } - } - - pub fn filter(&self) -> Filter { - match &self.project_tracking { - ProjectTracking::All { .. } => Filter::default(), - ProjectTracking::Allowed(ids) => Filter::new(ids.iter()), - } - } - pub fn alias(&self) -> [u8; 32] { let mut alias = [0u8; 32]; diff --git a/radicle-node/src/service/filter.rs b/radicle-node/src/service/filter.rs index 13273533..367e30dd 100644 --- a/radicle-node/src/service/filter.rs +++ b/radicle-node/src/service/filter.rs @@ -40,7 +40,7 @@ impl Filter { /// Create a new filter with the given items. /// /// Uses the iterator's size hint to determine the size of the filter. - pub fn new<'a>(ids: impl IntoIterator) -> Self { + pub fn new(ids: impl IntoIterator) -> Self { let iterator = ids.into_iter(); let (min, _) = iterator.size_hint(); let size = bloomy::bloom::optimal_bits(min, FILTER_FP_RATE) / 8; @@ -54,11 +54,16 @@ impl Filter { let mut bloom = BloomFilter::with_size(size); for id in iterator { - bloom.insert(id); + bloom.insert(&id); } Self(bloom) } + /// Empty filter with nothing set. + pub fn empty() -> Self { + Self(BloomFilter::from(vec![0x0; FILTER_SIZE_S])) + } + /// Size in bytes. pub fn size(&self) -> usize { self.0.bits() / 8 @@ -132,13 +137,13 @@ mod test { #[test] fn test_sizes() { let ids = arbitrary::vec::(3420); - let f = Filter::new(ids.iter().take(10)); + let f = Filter::new(ids.iter().cloned().take(10)); assert_eq!(f.size(), FILTER_SIZE_S); - let f = Filter::new(ids.iter().take(1000)); + let f = Filter::new(ids.iter().cloned().take(1000)); assert_eq!(f.size(), FILTER_SIZE_M); - let f = Filter::new(ids.iter()); + let f = Filter::new(ids.iter().cloned()); assert_eq!(f.size(), FILTER_SIZE_L); // Just checking that iterators over hash sets give correct size hints. diff --git a/radicle-node/src/service/tracking.rs b/radicle-node/src/service/tracking.rs new file mode 100644 index 00000000..da30f4b1 --- /dev/null +++ b/radicle-node/src/service/tracking.rs @@ -0,0 +1,38 @@ +mod store; + +use std::str::FromStr; + +pub use store::{Config, Error}; + +/// Tracking policy. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] +pub enum Policy { + /// The resource is tracked. + Track, + /// The resource is blocked. + Block, +} + +/// 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(()), + } + } +} diff --git a/radicle-node/src/service/tracking/schema.sql b/radicle-node/src/service/tracking/schema.sql new file mode 100644 index 00000000..0758761e --- /dev/null +++ b/radicle-node/src/service/tracking/schema.sql @@ -0,0 +1,32 @@ +-- +-- Service configuration schema. +-- + +-- Node tracking policy. +create table if not exists "node-policies" ( + -- Node ID. + "id" text primary key not null, + -- Node alias. May override the alias announced by the node. + "alias" text default '', + -- Tracking policy for this node. + "policy" text default 'track' + -- +) strict; + +-- Repository tracking policy. +create table if not exists "repo-policies" ( + -- Repository ID. + "id" text primary key not null, + -- Tracking scope for this repository. + -- + -- Valid values are: + -- + -- "trusted" track repository delegates and remotes in the `node-policies` table. + -- "delegates-only" only track repository delegates. + -- "all" track all remotes. + -- + "scope" text default 'trusted', + -- Tracking policy for this repository. + "policy" text default 'track' + -- +) strict; diff --git a/radicle-node/src/service/tracking/store.rs b/radicle-node/src/service/tracking/store.rs new file mode 100644 index 00000000..c0e656b1 --- /dev/null +++ b/radicle-node/src/service/tracking/store.rs @@ -0,0 +1,397 @@ +use std::path::Path; +use std::str::FromStr; +use std::{fmt, io}; + +use sqlite as sql; +use thiserror::Error; + +use crate::prelude::Id; +use crate::service::NodeId; + +use super::{Policy, Scope}; + +#[derive(Error, Debug)] +pub enum Error { + /// I/O error. + #[error("i/o error: {0}")] + Io(#[from] io::Error), + /// An Internal error. + #[error("internal error: {0}")] + Internal(#[from] sql::Error), +} + +pub type Alias = String; + +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, +} + +impl fmt::Debug for Config { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "Config(..)") + } +} + +impl Config { + const SCHEMA: &str = include_str!("schema.sql"); + + /// Open a policy store at the given path. Creates a new store if it + /// doesn't exist. + pub fn open>(path: P) -> Result { + let db = sql::Connection::open(path)?; + db.execute(Self::SCHEMA)?; + + Ok(Self { db }) + } + + /// Create a new in-memory address book. + pub fn memory() -> Result { + let db = sql::Connection::open(":memory:")?; + db.execute(Self::SCHEMA)?; + + Ok(Self { db }) + } + + /// Track a node. + pub fn track_node(&mut self, id: &NodeId, alias: Option<&str>) -> Result { + let mut stmt = self.db.prepare( + "INSERT INTO `node-policies` (id, alias) + VALUES (?1, ?2) + ON CONFLICT DO UPDATE + SET alias = ?2 WHERE alias != ?2", + )?; + + stmt.bind((1, id))?; + stmt.bind((2, alias.unwrap_or_default()))?; + stmt.next()?; + + Ok(self.db.change_count() > 0) + } + + /// Track a repository. + pub fn track_repo(&mut self, id: &Id, scope: Scope) -> Result { + let mut stmt = self.db.prepare( + "INSERT INTO `repo-policies` (id, scope) + VALUES (?1, ?2) + ON CONFLICT DO UPDATE + SET scope = ?2 WHERE scope != ?2", + )?; + + stmt.bind((1, id))?; + stmt.bind((2, scope))?; + stmt.next()?; + + Ok(self.db.change_count() > 0) + } + + /// Set a node's tracking policy. + pub fn set_node_policy(&mut self, id: &NodeId, policy: Policy) -> Result { + let mut stmt = self.db.prepare( + "INSERT INTO `node-policies` (id, policy) + VALUES (?1, ?2) + ON CONFLICT DO UPDATE + SET policy = ?2 WHERE policy != ?2", + )?; + + stmt.bind((1, id))?; + stmt.bind((2, policy))?; + stmt.next()?; + + Ok(self.db.change_count() > 0) + } + + /// Set a repository's tracking policy. + pub fn set_repo_policy(&mut self, id: &Id, policy: Policy) -> Result { + let mut stmt = self.db.prepare( + "INSERT INTO `repo-policies` (id, policy) + VALUES (?1, ?2) + ON CONFLICT DO UPDATE + SET policy = ?2 WHERE policy != ?2", + )?; + + stmt.bind((1, id))?; + stmt.bind((2, policy))?; + stmt.next()?; + + Ok(self.db.change_count() > 0) + } + + /// Untrack a node. + pub fn untrack_node(&mut self, id: &NodeId) -> Result { + let mut stmt = self + .db + .prepare("DELETE FROM `node-policies` WHERE id = ?")?; + + stmt.bind((1, id))?; + stmt.next()?; + + Ok(self.db.change_count() > 0) + } + + /// Untrack a repository. + pub fn untrack_repo(&mut self, id: &Id) -> Result { + let mut stmt = self + .db + .prepare("DELETE FROM `repo-policies` WHERE id = ?")?; + + stmt.bind((1, id))?; + stmt.next()?; + + Ok(self.db.change_count() > 0) + } + + /// Check if a node is tracked. + pub fn is_node_tracked(&self, id: &NodeId) -> Result { + Ok(matches!(self.node_entry(id)?, Some((_, Policy::Track)))) + } + + /// Check if a repository is tracked. + pub fn is_repo_tracked(&self, id: &Id) -> Result { + Ok(matches!(self.repo_entry(id)?, Some((_, Policy::Track)))) + } + + /// Get a node's tracking information. + pub fn node_entry(&self, id: &NodeId) -> Result, Policy)>, Error> { + let mut stmt = self + .db + .prepare("SELECT alias, policy FROM `node-policies` WHERE id = ?")?; + + stmt.bind((1, id))?; + + if let Some(Ok(row)) = stmt.into_iter().next() { + let alias = row.read::<&str, _>("alias"); + + return Ok(Some(( + if alias.is_empty() { + None + } else { + Some(alias.to_owned()) + }, + row.read::("policy"), + ))); + } + Ok(None) + } + + /// Get a repository's tracking information. + pub fn repo_entry(&self, id: &Id) -> Result, Error> { + let mut stmt = self + .db + .prepare("SELECT scope, policy FROM `repo-policies` WHERE id = ?")?; + + stmt.bind((1, id))?; + + if let Some(Ok(row)) = stmt.into_iter().next() { + return Ok(Some(( + row.read::("scope"), + row.read::("policy"), + ))); + } + Ok(None) + } + + /// Get node tracking entries. + pub fn node_entries(&self) -> Result>, Error> { + let mut stmt = self + .db + .prepare("SELECT id, alias FROM `node-policies`")? + .into_iter(); + let mut entries = Vec::new(); + + while let Some(Ok(row)) = stmt.next() { + let id = row.read("id"); + let alias = row.read::<&str, _>("alias"); + + entries.push((id, alias.to_owned())); + } + Ok(Box::new(entries.into_iter())) + } + + /// Get repository tracking entries. + pub fn repo_entries(&self) -> Result>, Error> { + let mut stmt = self + .db + .prepare("SELECT id, scope FROM `repo-policies`")? + .into_iter(); + let mut entries = Vec::new(); + + while let Some(Ok(row)) = stmt.next() { + let id = row.read("id"); + let scope = row.read("scope"); + + entries.push((id, scope)); + } + Ok(Box::new(entries.into_iter())) + } +} + +#[cfg(test)] +mod test { + use radicle::assert_matches; + + use super::*; + use crate::test::arbitrary; + + #[test] + fn test_track_and_untrack_node() { + let id = arbitrary::gen::(1); + let mut db = Config::open(":memory:").unwrap(); + + assert!(db.track_node(&id, Some("eve")).unwrap()); + assert!(db.is_node_tracked(&id).unwrap()); + assert!(!db.track_node(&id, Some("eve")).unwrap()); + assert!(db.untrack_node(&id).unwrap()); + assert!(!db.is_node_tracked(&id).unwrap()); + } + + #[test] + fn test_track_and_untrack_repo() { + let id = arbitrary::gen::(1); + let mut db = Config::open(":memory:").unwrap(); + + assert!(db.track_repo(&id, Scope::All).unwrap()); + assert!(db.is_repo_tracked(&id).unwrap()); + assert!(!db.track_repo(&id, Scope::All).unwrap()); + assert!(db.untrack_repo(&id).unwrap()); + assert!(!db.is_repo_tracked(&id).unwrap()); + } + + #[test] + fn test_node_entries() { + let ids = arbitrary::vec::(3); + let mut db = Config::open(":memory:").unwrap(); + + for id in &ids { + 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]); + } + + #[test] + fn test_repo_entries() { + let ids = arbitrary::vec::(3); + let mut db = Config::open(":memory:").unwrap(); + + for id in &ids { + 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]); + } + + #[test] + fn test_update_alias() { + let id = arbitrary::gen::(1); + let mut db = Config::open(":memory:").unwrap(); + + assert!(db.track_node(&id, Some("eve")).unwrap()); + assert_eq!( + db.node_entry(&id).unwrap().unwrap().0, + Some(String::from("eve")) + ); + assert!(db.track_node(&id, None).unwrap()); + assert_eq!(db.node_entry(&id).unwrap().unwrap().0, None); + assert!(!db.track_node(&id, None).unwrap()); + assert!(db.track_node(&id, Some("alice")).unwrap()); + assert_eq!( + db.node_entry(&id).unwrap().unwrap().0, + Some(String::from("alice")) + ); + } + + #[test] + fn test_update_scope() { + let id = arbitrary::gen::(1); + let mut db = Config::open(":memory:").unwrap(); + + assert!(db.track_repo(&id, Scope::All).unwrap()); + assert_eq!(db.repo_entry(&id).unwrap().unwrap().0, Scope::All); + assert!(db.track_repo(&id, Scope::DelegatesOnly).unwrap()); + assert_eq!(db.repo_entry(&id).unwrap().unwrap().0, Scope::DelegatesOnly); + } + + #[test] + fn test_repo_policy() { + let id = arbitrary::gen::(1); + let mut db = Config::open(":memory:").unwrap(); + + assert!(db.track_repo(&id, Scope::All).unwrap()); + assert_eq!(db.repo_entry(&id).unwrap().unwrap().1, Policy::Track); + assert!(db.set_repo_policy(&id, Policy::Block).unwrap()); + assert_eq!(db.repo_entry(&id).unwrap().unwrap().1, Policy::Block); + } + + #[test] + fn test_node_policy() { + let id = arbitrary::gen::(1); + let mut db = Config::open(":memory:").unwrap(); + + assert!(db.track_node(&id, None).unwrap()); + assert_eq!(db.node_entry(&id).unwrap().unwrap().1, Policy::Track); + assert!(db.set_node_policy(&id, Policy::Block).unwrap()); + assert_eq!(db.node_entry(&id).unwrap().unwrap().1, Policy::Block); + } +} diff --git a/radicle-node/src/test/peer.rs b/radicle-node/src/test/peer.rs index dea3f32b..e7b337b8 100644 --- a/radicle-node/src/test/peer.rs +++ b/radicle-node/src/test/peer.rs @@ -98,7 +98,17 @@ where let local_time = LocalTime::now(); let clock = RefClock::from(local_time); let routing = routing::Table::memory().unwrap(); - let service = Service::new(config, clock, routing, storage, addrs, signer, rng.clone()); + let tracking = tracking::Config::memory().unwrap(); + let service = Service::new( + config, + clock, + routing, + storage, + addrs, + tracking, + signer, + rng.clone(), + ); let ip = ip.into(); let local_addr = net::SocketAddr::new(ip, rng.u16(..)); diff --git a/radicle-node/src/tests.rs b/radicle-node/src/tests.rs index 50dbf033..d5d52e5e 100644 --- a/radicle-node/src/tests.rs +++ b/radicle-node/src/tests.rs @@ -343,10 +343,7 @@ fn test_inventory_pruning() { fn test_tracking() { let mut alice = Peer::config( "alice", - Config { - project_tracking: ProjectTracking::Allowed(HashSet::default()), - ..Config::default() - }, + Config::default(), [7, 7, 7, 7], MockStorage::empty(), address::Book::memory().unwrap(), @@ -362,7 +359,7 @@ fn test_tracking() { .map_err(client::handle::Error::from) .unwrap(); assert!(policy_change); - assert!(alice.config().is_tracking(&proj_id)); + assert!(alice.tracking().is_repo_tracked(&proj_id).unwrap()); let (sender, receiver) = chan::bounded(1); alice.command(Command::Untrack(proj_id, sender)); @@ -371,7 +368,7 @@ fn test_tracking() { .map_err(client::handle::Error::from) .unwrap(); assert!(policy_change); - assert!(!alice.config().is_tracking(&proj_id)); + assert!(!alice.tracking().is_repo_tracked(&proj_id).unwrap()); } #[test] @@ -559,9 +556,9 @@ fn test_refs_announcement_relay() { }; let bob_inv = bob.inventory().unwrap(); - alice.track(bob_inv[0]); - alice.track(bob_inv[1]); - alice.track(bob_inv[2]); + alice.track(&bob_inv[0], tracking::Scope::All).unwrap(); + alice.track(&bob_inv[1], tracking::Scope::All).unwrap(); + alice.track(&bob_inv[2], tracking::Scope::All).unwrap(); alice.connect_to(&bob); alice.connect_to(&eve); alice.receive(&eve.addr(), Message::Subscribe(Subscribe::all())); @@ -601,7 +598,7 @@ fn test_refs_announcement_no_subscribe() { let eve = Peer::new("eve", [9, 9, 9, 9], MockStorage::empty()); let id = arbitrary::gen(1); - alice.track(id); + alice.track(&id, tracking::Scope::All).unwrap(); alice.connect_to(&bob); alice.connect_to(&eve); alice.receive(&bob.addr(), bob.refs_announcement(id)); @@ -847,16 +844,6 @@ fn test_push_and_pull() { alice.command(service::Command::Connect(eve.addr())); bob.command(service::Command::Connect(eve.addr())); - let mut sim = Simulation::new( - LocalTime::now(), - alice.rng.clone(), - simulator::Options::default(), - ) - .initialize([&mut alice, &mut bob, &mut eve]); - - // Here we expect Alice to connect to Eve. - sim.run_while([&mut alice, &mut bob, &mut eve], |s| !s.is_settled()); - // Alice creates a new project. let (proj_id, _, _) = rad::init( &repo, @@ -876,7 +863,17 @@ fn test_push_and_pull() { let (sender, _) = chan::bounded(1); eve.command(service::Command::Track(proj_id, sender)); - // Neither of them have it in the beginning. + let mut sim = Simulation::new( + LocalTime::now(), + alice.rng.clone(), + simulator::Options::default(), + ) + .initialize([&mut alice, &mut bob, &mut eve]); + + // Here we expect Alice to connect to Eve. + sim.run_while([&mut alice, &mut bob, &mut eve], |s| !s.is_settled()); + + // Neither Eve nor Bob have Alice's project for now. assert!(eve.get(proj_id).unwrap().is_none()); assert!(bob.get(proj_id).unwrap().is_none()); diff --git a/radicle/src/sql.rs b/radicle/src/sql.rs index b6be543e..c89220cf 100644 --- a/radicle/src/sql.rs +++ b/radicle/src/sql.rs @@ -18,7 +18,7 @@ impl TryFrom<&Value> for Id { }), _ => Err(sql::Error { code: None, - message: None, + message: Some("sql: invalid type for id".to_owned()), }), } } @@ -44,7 +44,7 @@ impl TryFrom<&Value> for node::Features { Value::Integer(bits) => Ok(node::Features::from(*bits as u64)), _ => Err(sql::Error { code: None, - message: None, + message: Some("sql: invalid type for node features".to_owned()), }), } }