diff --git a/radicle-node/src/address_manager.rs b/radicle-node/src/address.rs similarity index 73% rename from radicle-node/src/address_manager.rs rename to radicle-node/src/address.rs index aa458663..52ea1916 100644 --- a/radicle-node/src/address_manager.rs +++ b/radicle-node/src/address.rs @@ -1,4 +1,8 @@ -use crate::address_book::Store; +mod store; +mod types; + +pub use store::*; +pub use types::*; #[derive(Debug)] pub struct AddressManager { diff --git a/radicle-node/src/address/schema.sql b/radicle-node/src/address/schema.sql new file mode 100644 index 00000000..4a6db837 --- /dev/null +++ b/radicle-node/src/address/schema.sql @@ -0,0 +1,37 @@ +-- +-- Address book SQL schema. +-- +create table if not exists "nodes" ( + -- Node ID. + "id" text primary key not null, + -- Node features. + "features" integer not null, + -- Node alias. + "alias" text default null, + -- Node announcement timestamp. + "timestamp" integer not null, + -- Local time at which we last attempted to connect to this node. + "last_attempt" integer default null, + -- Local time at which we successfully connected to this node. + "last_success" integer default null + -- +) strict; + +create table if not exists "addresses" ( + -- Node ID. + "node" text not null references "nodes" ("id"), + -- Address type. + "type" text not null, + -- Address value. + "value" text not null, + -- Where we got this address from. + "source" text not null, + -- When this address was announced. + "timestamp" integer not null, + -- Nb. This constraint allows more than one node to share the same address. + -- This is useful in circumstances when a node wants to rotate its key, but + -- remain reachable at the same address. The old entry will eventually be + -- pruned. + unique ("node", "type", "value") + -- +) strict; diff --git a/radicle-node/src/address/store.rs b/radicle-node/src/address/store.rs new file mode 100644 index 00000000..8ea2d715 --- /dev/null +++ b/radicle-node/src/address/store.rs @@ -0,0 +1,497 @@ +use std::path::Path; +use std::str::FromStr; +use std::{fmt, io}; + +use radicle::node; +use sqlite as sql; +use thiserror::Error; + +use crate::address::types; +use crate::address::{KnownAddress, Source}; +use crate::clock::Timestamp; +use crate::prelude::Address; +use crate::service::NodeId; +use crate::wire::message::AddressType; + +#[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), +} + +/// A file-backed address book. +pub struct Book { + db: sql::Connection, +} + +impl fmt::Debug for Book { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "Cache(..)") + } +} + +impl Book { + const SCHEMA: &str = include_str!("schema.sql"); + + /// Open an address book at the given path. Creates a new address book 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 }) + } +} + +impl Store for Book { + fn get(&self, node: &NodeId) -> Result, Error> { + let row = self + .db + .prepare("SELECT features, alias, timestamp FROM nodes WHERE id = ?")? + .bind(1, node)? + .into_cursor() + .next(); + + if let Some(Ok(row)) = row { + let features = row.get::(0); + let alias = row.get::(1); + let timestamp = row.get::(2) as Timestamp; + let mut addrs = Vec::new(); + + let mut stmt = self + .db + .prepare("SELECT type, value, source FROM addresses WHERE node = ?")? + .bind(1, node)? + .into_cursor(); + + while let Some(Ok(row)) = stmt.next() { + let _typ = row.get::(0); + let addr = row.get::(1); + let source = row.get::(2); + + addrs.push(KnownAddress { + addr, + source, + last_success: None, + last_attempt: None, + }); + } + + Ok(Some(types::Node { + features, + alias, + timestamp, + addrs, + })) + } else { + Ok(None) + } + } + + fn len(&self) -> Result { + let row = self + .db + .prepare("SELECT COUNT(*) FROM addresses")? + .into_cursor() + .next() + .unwrap() + .unwrap(); + let count = row.get::(0) as usize; + + Ok(count) + } + + fn insert( + &mut self, + node: &NodeId, + features: node::Features, + alias: &str, + timestamp: Timestamp, + addrs: impl IntoIterator, + ) -> Result { + self.db + .prepare( + "INSERT INTO nodes (id, features, alias, timestamp) + VALUES (?1, ?2, ?3, ?4) + ON CONFLICT DO UPDATE + SET features = ?2, alias = ?3, timestamp = ?4 + WHERE timestamp < ?4", + )? + .bind(1, node)? + .bind(2, features)? + .bind(3, alias)? + .bind(4, timestamp as i64)? + .next()?; + + for addr in addrs { + self.db + .prepare( + "INSERT INTO addresses (node, type, value, source, timestamp) + VALUES (?1, ?2, ?3, ?4, ?5) + ON CONFLICT DO UPDATE + SET timestamp = ?5 + WHERE timestamp < ?5", + )? + .bind(1, node)? + .bind(2, AddressType::from(&addr.addr))? + .bind(3, addr.addr)? + .bind(4, addr.source)? + .bind(5, timestamp as i64)? + .next()?; + } + Ok(self.db.change_count() > 0) + } + + fn remove(&mut self, node: &NodeId) -> Result { + self.db + .prepare("DELETE FROM nodes WHERE id = ?")? + .bind(1, node)? + .next()?; + self.db + .prepare("DELETE FROM addresses WHERE node = ?")? + .bind(1, node)? + .next()?; + + Ok(self.db.change_count() > 0) + } + + fn entries(&self) -> Result>, Error> { + let mut stmt = self + .db + .prepare("SELECT node, type, value, source FROM addresses ORDER BY node")? + .into_cursor(); + let mut entries = Vec::new(); + + while let Some(Ok(row)) = stmt.next() { + let node = row.get(0); + let _typ = row.get::(1); + let addr = row.get::(2); + let source = row.get::(3); + + entries.push(( + node, + KnownAddress { + addr, + source, + last_success: None, + last_attempt: None, + }, + )); + } + Ok(Box::new(entries.into_iter())) + } +} + +/// Address store. +/// +/// Used to store node addresses and metadata. +pub trait Store { + /// Get a known peer address. + fn get(&self, id: &NodeId) -> Result, Error>; + /// Insert a node with associated addresses into the store. + /// + /// Returns `true` if the node or addresses were updated, and `false` otherwise. + fn insert( + &mut self, + node: &NodeId, + features: node::Features, + alias: &str, + timestamp: Timestamp, + addrs: impl IntoIterator, + ) -> Result; + /// Remove an address from the store. + fn remove(&mut self, id: &NodeId) -> Result; + /// Returns the number of addresses. + fn len(&self) -> Result; + /// Returns true if there are no addresses. + fn is_empty(&self) -> Result { + self.len().map(|l| l == 0) + } + /// Get the address entries in the store. + fn entries(&self) -> Result>, Error>; +} + +impl sql::ValueInto for Address { + fn into(value: &sql::Value) -> Option { + match value { + sql::Value::String(s) => Address::from_str(s.as_str()).ok(), + _ => None, + } + } +} + +impl sql::Bindable for Address { + fn bind(self, stmt: &mut sql::Statement<'_>, i: usize) -> sql::Result<()> { + self.to_string().bind(stmt, i) + } +} + +impl sql::ValueInto for Source { + fn into(value: &sql::Value) -> Option { + match value { + sql::Value::String(s) => match s.as_str() { + "dns" => Some(Source::Dns), + "peer" => Some(Source::Peer), + "imported" => Some(Source::Imported), + _ => None, + }, + _ => None, + } + } +} + +impl sql::Bindable for Source { + fn bind(self, stmt: &mut sql::Statement<'_>, i: usize) -> sql::Result<()> { + match self { + Self::Dns => "dns".bind(stmt, i), + Self::Peer => "peer".bind(stmt, i), + Self::Imported => "imported".bind(stmt, i), + } + } +} + +impl sql::ValueInto for AddressType { + fn into(value: &sql::Value) -> Option { + match value { + sql::Value::String(s) => match s.as_str() { + "ipv4" => Some(AddressType::Ipv4), + "ipv6" => Some(AddressType::Ipv6), + "hostname" => Some(AddressType::Hostname), + "onion" => Some(AddressType::Onion), + _ => None, + }, + _ => None, + } + } +} + +impl sql::Bindable for AddressType { + fn bind(self, stmt: &mut sql::Statement<'_>, i: usize) -> sql::Result<()> { + match self { + Self::Ipv4 => "ipv4".bind(stmt, i), + Self::Ipv6 => "ipv6".bind(stmt, i), + Self::Hostname => "hostname".bind(stmt, i), + Self::Onion => "onion".bind(stmt, i), + } + } +} + +#[cfg(test)] +mod test { + use std::net; + + use super::*; + use crate::test::arbitrary; + use crate::LocalTime; + + #[test] + fn test_empty() { + let tmp = tempfile::tempdir().unwrap(); + let path = tmp.path().join("cache"); + let cache = Book::open(&path).unwrap(); + + assert!(cache.is_empty().unwrap()); + } + + #[test] + fn test_get_none() { + let alice = arbitrary::gen::(1); + let cache = Book::memory().unwrap(); + let result = cache.get(&alice).unwrap(); + + assert!(result.is_none()); + } + + #[test] + fn test_remove_nothing() { + let alice = arbitrary::gen::(1); + let mut cache = Book::memory().unwrap(); + let removed = cache.remove(&alice).unwrap(); + + assert!(!removed); + } + + #[test] + fn test_insert_and_get() { + let alice = arbitrary::gen::(1); + let mut cache = Book::memory().unwrap(); + let features = node::Features::SEED; + let timestamp = LocalTime::now().as_secs(); + + let ka = KnownAddress { + addr: net::SocketAddr::from(([4, 4, 4, 4], 8776)).into(), + source: Source::Peer, + last_success: None, + last_attempt: None, + }; + let inserted = cache + .insert(&alice, features, "alice", timestamp, [ka.clone()]) + .unwrap(); + assert!(inserted); + + let node = cache.get(&alice).unwrap().unwrap(); + + assert_eq!(node.features, features); + assert_eq!(node.timestamp, timestamp); + assert_eq!(node.alias.as_str(), "alice"); + assert_eq!(node.addrs, vec![ka]); + } + + #[test] + fn test_insert_duplicate() { + let alice = arbitrary::gen::(1); + let mut cache = Book::memory().unwrap(); + let features = node::Features::SEED; + let timestamp = LocalTime::now().as_secs(); + + let ka = KnownAddress { + addr: net::SocketAddr::from(([4, 4, 4, 4], 8776)).into(), + source: Source::Peer, + last_success: None, + last_attempt: None, + }; + let inserted = cache + .insert(&alice, features, "alice", timestamp, [ka.clone()]) + .unwrap(); + assert!(inserted); + + let inserted = cache + .insert(&alice, features, "alice", timestamp, [ka]) + .unwrap(); + assert!(!inserted); + + assert_eq!(cache.len().unwrap(), 1); + } + + #[test] + fn test_insert_and_update() { + let alice = arbitrary::gen::(1); + let mut cache = Book::memory().unwrap(); + let timestamp = LocalTime::now().as_secs(); + let features = node::Features::SEED; + let ka = KnownAddress { + addr: net::SocketAddr::from(([4, 4, 4, 4], 8776)).into(), + source: Source::Peer, + last_success: None, + last_attempt: None, + }; + + let updated = cache + .insert(&alice, features, "alice", timestamp, [ka.clone()]) + .unwrap(); + assert!(updated); + + let updated = cache + .insert(&alice, features, "~alice~", timestamp, []) + .unwrap(); + assert!(!updated, "Can't update using the same timestamp"); + + let updated = cache + .insert(&alice, features, "~alice~", timestamp - 1, []) + .unwrap(); + assert!(!updated, "Can't update using a smaller timestamp"); + + let node = cache.get(&alice).unwrap().unwrap(); + assert_eq!(node.alias, "alice"); + assert_eq!(node.timestamp, timestamp); + + let updated = cache + .insert(&alice, features, "~alice~", timestamp + 1, []) + .unwrap(); + assert!(updated, "Can update with a larger timestamp"); + + let updated = cache + .insert(&alice, node::Features::NONE, "~alice~", timestamp + 2, []) + .unwrap(); + assert!(updated); + + let node = cache.get(&alice).unwrap().unwrap(); + assert_eq!(node.features, node::Features::NONE); + assert_eq!(node.alias, "~alice~"); + assert_eq!(node.timestamp, timestamp + 2); + assert_eq!(node.addrs, vec![ka]); + } + + #[test] + fn test_insert_and_remove() { + let alice = arbitrary::gen::(1); + let bob = arbitrary::gen::(1); + let mut cache = Book::memory().unwrap(); + let timestamp = LocalTime::now().as_secs(); + let features = node::Features::SEED; + + for addr in [ + ([4, 4, 4, 4], 8776), + ([7, 7, 7, 7], 8776), + ([9, 9, 9, 9], 8776), + ] { + let ka = KnownAddress { + addr: net::SocketAddr::from(addr).into(), + source: Source::Peer, + last_success: None, + last_attempt: None, + }; + cache + .insert(&alice, features, "alice", timestamp, [ka.clone()]) + .unwrap(); + cache + .insert(&bob, features, "bob", timestamp, [ka]) + .unwrap(); + } + assert_eq!(cache.len().unwrap(), 6); + + let removed = cache.remove(&alice).unwrap(); + assert!(removed); + assert_eq!(cache.len().unwrap(), 3); + + let removed = cache.remove(&bob).unwrap(); + assert!(removed); + assert_eq!(cache.len().unwrap(), 0); + } + + #[test] + fn test_entries() { + let ids = arbitrary::vec::(16); + let rng = fastrand::Rng::new(); + let mut cache = Book::memory().unwrap(); + let mut expected = Vec::new(); + let timestamp = LocalTime::now().as_secs(); + let features = node::Features::SEED; + + for id in ids { + let ip = rng.u32(..); + let addr = net::SocketAddr::from((net::Ipv4Addr::from(ip), rng.u16(..))); + let ka = KnownAddress { + addr: addr.into(), + source: Source::Dns, + // TODO: Test times as well. + last_success: None, + last_attempt: None, + }; + expected.push((id, ka.clone())); + cache + .insert(&id, features, "alias", timestamp, [ka]) + .unwrap(); + } + + let mut actual = cache.entries().unwrap().into_iter().collect::>(); + + actual.sort_by_key(|(i, _)| *i); + expected.sort_by_key(|(i, _)| *i); + + assert_eq!(cache.len().unwrap(), actual.len()); + assert_eq!(actual, expected); + } +} diff --git a/radicle-node/src/address/types.rs b/radicle-node/src/address/types.rs new file mode 100644 index 00000000..bf1b8564 --- /dev/null +++ b/radicle-node/src/address/types.rs @@ -0,0 +1,135 @@ +use std::ops::{Deref, DerefMut}; + +use nonempty::NonEmpty; +use radicle::node; + +use crate::clock::Timestamp; +use crate::collections::HashMap; +use crate::service::message::Address; +use crate::LocalTime; + +/// A map with the ability to randomly select values. +#[derive(Debug)] +pub struct AddressBook { + inner: HashMap, + rng: fastrand::Rng, +} + +impl AddressBook { + /// Create a new address book. + pub fn new(rng: fastrand::Rng) -> Self { + Self { + inner: HashMap::with_hasher(rng.clone().into()), + rng, + } + } + + /// Pick a random value in the book. + pub fn sample(&self) -> Option<(&K, &V)> { + self.sample_with(|_, _| true) + } + + /// Pick a random value in the book matching a predicate. + pub fn sample_with(&self, mut predicate: impl FnMut(&K, &V) -> bool) -> Option<(&K, &V)> { + if let Some(pairs) = NonEmpty::from_vec( + self.inner + .iter() + .filter(|(k, v)| predicate(*k, *v)) + .collect(), + ) { + let ix = self.rng.usize(..pairs.len()); + let pair = pairs[ix]; // Can't fail. + + Some(pair) + } else { + None + } + } + + /// Cycle through the keys at random. The random cycle repeats ad-infintum. + pub fn cycle(&self) -> impl Iterator { + self.shuffled().map(|(k, _)| k).cycle() + } + + /// Return a shuffled iterator over the keys. + pub fn shuffled(&self) -> std::vec::IntoIter<(&K, &V)> { + let mut keys = self.inner.iter().collect::>(); + self.rng.shuffle(&mut keys); + + keys.into_iter() + } +} + +impl Deref for AddressBook { + type Target = HashMap; + + fn deref(&self) -> &Self::Target { + &self.inner + } +} + +impl DerefMut for AddressBook { + fn deref_mut(&mut self) -> &mut Self::Target { + &mut self.inner + } +} + +/// Node public data. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Node { + /// Advertized alias. + pub alias: String, + /// Advertized features. + pub features: node::Features, + /// Advertized addresses + pub addrs: Vec, + /// When this data was published. + pub timestamp: Timestamp, +} + +/// A known address. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct KnownAddress { + /// Network address. + pub addr: Address, + /// Address of the peer who sent us this address. + pub source: Source, + /// Last time this address was used to successfully connect to a peer. + pub last_success: Option, + /// Last time this address was tried. + pub last_attempt: Option, +} + +impl KnownAddress { + /// Create a new known address. + pub fn new(addr: Address, source: Source) -> Self { + Self { + addr, + source, + last_success: None, + last_attempt: None, + } + } +} + +/// Address source. Specifies where an address originated from. +#[derive(Debug, Copy, Clone, PartialEq, Eq)] +pub enum Source { + /// An address that was shared by another peer. + Peer, + /// An address that came from a DNS seed. + Dns, + /// An address that came from some source external to the system, eg. + /// specified by the user or added directly to the address manager. + Imported, +} + +impl std::fmt::Display for Source { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Peer => write!(f, "Peer"), + Self::Dns => write!(f, "DNS"), + Self::Imported => write!(f, "Imported"), + } + } +} diff --git a/radicle-node/src/address_book.rs b/radicle-node/src/address_book.rs deleted file mode 100644 index 7fe0a96d..00000000 --- a/radicle-node/src/address_book.rs +++ /dev/null @@ -1,468 +0,0 @@ -use std::io::Seek; -use std::ops::{Deref, DerefMut}; -use std::path::Path; -use std::{fs, io, net}; - -use crate::collections::HashMap; -use crate::LocalTime; -use nonempty::NonEmpty; -use serde::{Deserialize, Serialize}; - -/// A map with the ability to randomly select values. -#[derive(Debug)] -pub struct AddressBook { - inner: HashMap, - rng: fastrand::Rng, -} - -impl AddressBook { - /// Create a new address book. - pub fn new(rng: fastrand::Rng) -> Self { - Self { - inner: HashMap::with_hasher(rng.clone().into()), - rng, - } - } - - /// Pick a random value in the book. - pub fn sample(&self) -> Option<(&K, &V)> { - self.sample_with(|_, _| true) - } - - /// Pick a random value in the book matching a predicate. - pub fn sample_with(&self, mut predicate: impl FnMut(&K, &V) -> bool) -> Option<(&K, &V)> { - if let Some(pairs) = NonEmpty::from_vec( - self.inner - .iter() - .filter(|(k, v)| predicate(*k, *v)) - .collect(), - ) { - let ix = self.rng.usize(..pairs.len()); - let pair = pairs[ix]; // Can't fail. - - Some(pair) - } else { - None - } - } - - /// Cycle through the keys at random. The random cycle repeats ad-infintum. - pub fn cycle(&self) -> impl Iterator { - self.shuffled().map(|(k, _)| k).cycle() - } - - /// Return a shuffled iterator over the keys. - pub fn shuffled(&self) -> std::vec::IntoIter<(&K, &V)> { - let mut keys = self.inner.iter().collect::>(); - self.rng.shuffle(&mut keys); - - keys.into_iter() - } -} - -impl Deref for AddressBook { - type Target = HashMap; - - fn deref(&self) -> &Self::Target { - &self.inner - } -} - -impl DerefMut for AddressBook { - fn deref_mut(&mut self) -> &mut Self::Target { - &mut self.inner - } -} - -/// A known address. -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -pub struct KnownAddress { - /// Network address. - pub addr: net::SocketAddr, - /// Address of the peer who sent us this address. - pub source: Source, - /// Last time this address was used to successfully connect to a peer. - #[serde(with = "local_time")] - pub last_success: Option, - /// Last time this address was sampled. - #[serde(with = "local_time")] - pub last_sampled: Option, - /// Last time this address was tried. - #[serde(with = "local_time")] - pub last_attempt: Option, - /// Last time this peer was seen alive. - #[serde(with = "local_time")] - pub last_active: Option, -} - -impl KnownAddress { - /// Create a new known address. - pub fn new(addr: net::SocketAddr, source: Source, last_active: Option) -> Self { - Self { - addr, - source, - last_success: None, - last_attempt: None, - last_sampled: None, - last_active, - } - } -} - -/// Address source. Specifies where an address originated from. -#[derive(Debug, Copy, Clone, PartialEq, Eq, Serialize, Deserialize)] -pub enum Source { - /// An address that was shared by another peer. - Peer(net::SocketAddr), - /// An address that came from a DNS seed. - Dns, - /// An address that came from some source external to the system, eg. - /// specified by the user or added directly to the address manager. - Imported, -} - -impl std::fmt::Display for Source { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - match self { - Self::Peer(addr) => write!(f, "{}", addr), - Self::Dns => write!(f, "DNS"), - Self::Imported => write!(f, "Imported"), - } - } -} - -/// A file-backed address cache. -#[derive(Debug)] -pub struct Cache { - addrs: std::collections::HashMap, - file: fs::File, -} - -impl Cache { - /// Open an existing cache. - pub fn open>(path: P) -> io::Result { - fs::OpenOptions::new() - .read(true) - .write(true) - .open(path) - .and_then(Self::from) - } - - /// Create a new cache. - pub fn create>(path: P) -> io::Result { - use std::collections::HashMap; - - let file = fs::OpenOptions::new() - .create_new(true) - .write(true) - .open(path)?; - - Ok(Self { - file, - addrs: HashMap::new(), - }) - } - - /// Create a new cache from a file. - pub fn from(mut file: fs::File) -> io::Result { - use std::collections::HashMap; - - let bytes = file.seek(io::SeekFrom::End(0))?; - let addrs = if bytes == 0 { - HashMap::new() - } else { - file.rewind()?; - serde_json::from_reader(&file)? - }; - - Ok(Self { file, addrs }) - } -} - -impl Store for Cache { - fn get_mut(&mut self, ip: &net::IpAddr) -> Option<&mut KnownAddress> { - self.addrs.get_mut(ip) - } - - fn get(&self, ip: &net::IpAddr) -> Option<&KnownAddress> { - self.addrs.get(ip) - } - - fn remove(&mut self, ip: &net::IpAddr) -> Option { - self.addrs.remove(ip) - } - - fn insert(&mut self, ip: net::IpAddr, ka: KnownAddress) -> bool { - as Store>::insert(&mut self.addrs, ip, ka) - } - - fn iter<'a>(&'a self) -> Box + 'a> { - Box::new(self.addrs.iter()) - } - - fn clear(&mut self) { - self.addrs.clear() - } - - fn len(&self) -> usize { - self.addrs.len() - } - - fn flush<'a>(&mut self) -> io::Result<()> { - use io::Write; - - let peers = serde_json::to_value(&self.addrs)?; - let s = serde_json::to_string(&peers)?; - - self.file.set_len(0)?; - self.file.seek(io::SeekFrom::Start(0))?; - self.file.write_all(s.as_bytes())?; - self.file.write_all(&[b'\n'])?; - self.file.sync_data()?; - - Ok(()) - } -} - -/// Address store. -/// -/// Used to store peer addresses and metadata. -pub trait Store { - /// Get a known peer address. - fn get(&self, ip: &net::IpAddr) -> Option<&KnownAddress>; - - /// Get a known peer address mutably. - fn get_mut(&mut self, ip: &net::IpAddr) -> Option<&mut KnownAddress>; - - /// Insert a *new* address into the store. Returns `true` if the address was inserted, - /// or `false` if it was already known. - fn insert(&mut self, ip: net::IpAddr, ka: KnownAddress) -> bool; - - /// Remove an address from the store. - fn remove(&mut self, ip: &net::IpAddr) -> Option; - - /// Return an iterator over the known addresses. - fn iter<'a>(&'a self) -> Box + 'a>; - - /// Returns the number of addresses. - fn len(&self) -> usize; - - /// Returns true if there are no addresses. - fn is_empty(&self) -> bool { - self.len() == 0 - } - - /// Seed the peer store with addresses. - /// Fails if *none* of the seeds could be resolved to addresses. - fn seed( - &mut self, - seeds: impl Iterator, - source: Source, - ) -> io::Result<()> { - let mut error = None; - let mut success = false; - - for seed in seeds { - match seed.to_socket_addrs() { - Ok(addrs) => { - success = true; - for addr in addrs { - self.insert(addr.ip(), KnownAddress::new(addr, source, None)); - } - } - Err(err) => error = Some(err), - } - } - - if success { - return Ok(()); - } - if let Some(err) = error { - return Err(io::Error::new( - io::ErrorKind::Other, - format!("seeds failed to resolve: {}", err), - )); - } - Ok(()) - } - - /// Clears the store of all addresses. - fn clear(&mut self); - - /// Flush data to permanent storage. - fn flush(&mut self) -> io::Result<()>; -} - -/// Implementation of [`Store`] for [`std::collections::HashMap`]. -impl Store for std::collections::HashMap { - fn get_mut(&mut self, ip: &net::IpAddr) -> Option<&mut KnownAddress> { - self.get_mut(ip) - } - - fn get(&self, ip: &net::IpAddr) -> Option<&KnownAddress> { - self.get(ip) - } - - fn remove(&mut self, ip: &net::IpAddr) -> Option { - self.remove(ip) - } - - fn insert(&mut self, ip: net::IpAddr, ka: KnownAddress) -> bool { - use std::collections::hash_map::Entry; - - match self.entry(ip) { - Entry::Vacant(v) => { - v.insert(ka); - } - Entry::Occupied(_) => return false, - } - true - } - - fn iter<'a>(&'a self) -> Box + 'a> { - Box::new(self.iter()) - } - - fn clear(&mut self) { - self.clear() - } - - fn len(&self) -> usize { - self.len() - } - - fn flush(&mut self) -> std::io::Result<()> { - Ok(()) - } -} - -/// Implementation of [`Store`] for [`crate::collections::HashMap`]. -impl Store for crate::collections::HashMap { - fn get_mut(&mut self, ip: &net::IpAddr) -> Option<&mut KnownAddress> { - self.get_mut(ip) - } - - fn get(&self, ip: &net::IpAddr) -> Option<&KnownAddress> { - self.get(ip) - } - - fn remove(&mut self, ip: &net::IpAddr) -> Option { - self.remove(ip) - } - - fn insert(&mut self, ip: net::IpAddr, ka: KnownAddress) -> bool { - use std::collections::hash_map::Entry; - - match self.entry(ip) { - Entry::Vacant(v) => { - v.insert(ka); - } - Entry::Occupied(_) => return false, - } - true - } - - fn iter<'a>(&'a self) -> Box + 'a> { - Box::new(self.iter()) - } - - fn clear(&mut self) { - self.clear() - } - - fn len(&self) -> usize { - self.len() - } - - fn flush(&mut self) -> std::io::Result<()> { - Ok(()) - } -} - -mod local_time { - use super::LocalTime; - use serde::{Deserialize, Deserializer, Serializer}; - - pub fn deserialize<'de, D>(deserializer: D) -> Result, D::Error> - where - D: Deserializer<'de>, - { - let value: Option = Deserialize::deserialize(deserializer)?; - - if let Some(value) = value { - Ok(Some(LocalTime::from_secs(value))) - } else { - Ok(None) - } - } - - pub fn serialize(value: &Option, serializer: S) -> Result - where - S: Serializer, - { - if let Some(local_time) = value { - serializer.serialize_u64(local_time.as_secs()) - } else { - serializer.serialize_none() - } - } -} - -#[cfg(test)] -mod test { - use super::*; - - #[test] - fn test_empty() { - let tmp = tempfile::tempdir().unwrap(); - let path = tmp.path().join("cache"); - - Cache::create(&path).unwrap(); - let cache = Cache::open(&path).unwrap(); - - assert!(cache.is_empty()); - } - - #[test] - fn test_save_and_load() { - let tmp = tempfile::tempdir().unwrap(); - let path = tmp.path().join("cache"); - let mut expected = Vec::new(); - - { - let mut cache = Cache::create(&path).unwrap(); - - for i in 32..48 { - let ip = net::IpAddr::from([127, 0, 0, i]); - let addr = net::SocketAddr::from((ip, 8333)); - let ka = KnownAddress { - addr, - source: Source::Dns, - last_success: Some(LocalTime::from_secs(i as u64)), - last_sampled: Some(LocalTime::from_secs((i + 1) as u64)), - last_attempt: None, - last_active: None, - }; - cache.insert(ip, ka); - } - cache.flush().unwrap(); - - for (ip, ka) in cache.iter() { - expected.push((*ip, ka.clone())); - } - } - - { - let cache = Cache::open(&path).unwrap(); - let mut actual = cache - .iter() - .map(|(i, ka)| (*i, ka.clone())) - .collect::>(); - - actual.sort_by_key(|(i, _)| *i); - expected.sort_by_key(|(i, _)| *i); - - assert_eq!(actual, expected); - } - } -} diff --git a/radicle-node/src/client.rs b/radicle-node/src/client.rs index 7e3d03f8..cf8f8200 100644 --- a/radicle-node/src/client.rs +++ b/radicle-node/src/client.rs @@ -5,12 +5,11 @@ use nakamoto_net::{LocalTime, Reactor}; use thiserror::Error; use crate::clock::RefClock; -use crate::collections::HashMap; use crate::profile::Profile; -use crate::service; use crate::service::routing; use crate::transport::Transport; use crate::wire::Wire; +use crate::{address, service}; pub mod handle; @@ -18,6 +17,8 @@ pub mod handle; pub const NODE_DIR: &str = "node"; /// Filename of routing table database under [`NODE_DIR`]. pub const ROUTING_DB_FILE: &str = "routing.db"; +/// Filename of address database under [`NODE_DIR`]. +pub const ADDRESS_DB_FILE: &str = "addresses.db"; /// A client error. #[derive(Error, Debug)] @@ -25,6 +26,9 @@ pub enum Error { /// A routing database error. #[error("routing database error: {0}")] Routing(#[from] routing::Error), + /// An address database error. + #[error("address database error: {0}")] + Addresses(#[from] address::Error), /// An I/O error. #[error("i/o error: {0}")] Io(#[from] io::Error), @@ -100,7 +104,8 @@ impl Client { let time = LocalTime::now(); let storage = self.profile.storage; let signer = self.profile.signer; - let addresses = HashMap::with_hasher(rng.clone().into()); + let addresses = + address::Book::open(self.profile.home.join(NODE_DIR).join(ADDRESS_DB_FILE))?; let routing = routing::Table::open(self.profile.home.join(NODE_DIR).join(ROUTING_DB_FILE))?; log::info!("Initializing client ({:?})..", network); diff --git a/radicle-node/src/lib.rs b/radicle-node/src/lib.rs index bfbb9298..f97a52d2 100644 --- a/radicle-node/src/lib.rs +++ b/radicle-node/src/lib.rs @@ -1,6 +1,5 @@ #![allow(dead_code)] -pub mod address_book; -pub mod address_manager; +pub mod address; pub mod client; pub mod clock; pub mod control; diff --git a/radicle-node/src/service.rs b/radicle-node/src/service.rs index 8388373c..d6f1e48a 100644 --- a/radicle-node/src/service.rs +++ b/radicle-node/src/service.rs @@ -8,7 +8,7 @@ pub mod routing; use std::collections::BTreeMap; use std::ops::{Deref, DerefMut}; use std::sync::Arc; -use std::{fmt, net, net::IpAddr}; +use std::{fmt, net, net::IpAddr, str}; use crossbeam_channel as chan; use fastrand::Rng; @@ -17,11 +17,11 @@ use nakamoto::{LocalDuration, LocalTime}; use nakamoto_net as nakamoto; use nakamoto_net::Link; use nonempty::NonEmpty; +use radicle::node::Features; use radicle::storage::ReadStorage; -use crate::address_book; -use crate::address_book::AddressBook; -use crate::address_manager::AddressManager; +use crate::address; +use crate::address::AddressBook; use crate::clock::{RefClock, Timestamp}; use crate::crypto; use crate::crypto::{Signer, Verified}; @@ -172,6 +172,8 @@ pub struct Service { storage: S, /// Network routing table. Keeps track of where projects are located. routing: R, + /// Node address manager. + addresses: A, /// State relating to gossip. gossip: Gossip, /// Peer sessions, currently or recently connected. @@ -182,8 +184,6 @@ pub struct Service { clock: RefClock, /// Interface to the I/O reactor. reactor: Reactor, - /// Peer address manager. - addrmgr: AddressManager, /// Source of entropy. rng: Rng, /// Whether our local inventory no long represents what we have announced to the network. @@ -218,7 +218,7 @@ where impl Service where R: routing::Store, - A: address_book::Store, + A: address::Store, S: WriteStorage + 'static, G: crypto::Signer, { @@ -231,19 +231,19 @@ where signer: G, rng: Rng, ) -> Self { - let addrmgr = AddressManager::new(addresses); let sessions = Sessions::new(rng.clone()); let network = config.network; Self { config, storage, - addrmgr, + addresses, signer, rng, clock, routing, gossip: Gossip::default(), + // FIXME: This should be loaded from the address store. peers: BTreeMap::new(), reactor: Reactor::new(network), sessions, @@ -607,22 +607,23 @@ where } let Announcement { node, message, .. } = announcement; let now = self.clock.local_time(); + let timestamp = message.timestamp(); + let relay = self.config.relay; // Don't allow messages from too far in the future. - if message.timestamp().saturating_sub(now.as_secs()) > MAX_TIME_DELTA.as_secs() { - return Err(SessionError::InvalidTimestamp(message.timestamp())); + if timestamp.saturating_sub(now.as_secs()) > MAX_TIME_DELTA.as_secs() { + return Err(SessionError::InvalidTimestamp(timestamp)); } match message { AnnouncementMessage::Inventory(message) => { - let peer = self.peers.entry(*node).or_insert_with(Peer::default); - let relay = self.config.relay; - // Discard inventory messages we've already seen, otherwise update // out last seen time. - if message.timestamp > peer.last_message { - peer.last_message = message.timestamp; + let peer = self.peers.entry(*node).or_insert_with(Peer::default); + if timestamp > peer.last_message { + peer.last_message = timestamp; } else { + debug!("Ignoring stale announcement from {node}"); return Ok(false); } @@ -639,16 +640,13 @@ where // origin of it. return Ok(false); } - - if relay { - return Ok(true); - } + return Ok(relay); } // Process a peer inventory update announcement by (maybe) fetching. AnnouncementMessage::Refs(message) => { - // FIXME: Check message timestamp. // TODO: Buffer/throttle fetches. // TODO: Check that we're tracking this user as well. + // FIXME: Discard old messages. if self.config.is_tracking(&message.id) { // TODO: Check refs to see if we should try to fetch or not. // FIXME: This code is wrong: we shouldn't be fetching from the connected peer, @@ -663,13 +661,52 @@ where }); if is_updated { - return Ok(true); + return Ok(relay); } } } - AnnouncementMessage::Node(_message) => { - // FIXME: Check message timestamp. - log::warn!("Node announcement handling is not implemented"); + AnnouncementMessage::Node(NodeAnnouncement { + features, + alias, + addresses, + timestamp, + }) => { + // FIXME: Discard old messages. + // If this node isn't a seed, we're not interested in adding it + // to our address book, but other nodes may be, so we relay the message anyway. + if !features.has(Features::SEED) { + return Ok(relay); + } + + let alias = match str::from_utf8(alias) { + Ok(s) => s, + Err(e) => { + warn!("Dropping node announcement from {node}: invalid alias: {e}"); + return Ok(false); + } + }; + + match self.addresses.insert( + node, + *features, + alias, + *timestamp, + addresses + .iter() + .map(|a| address::KnownAddress::new(a.clone(), address::Source::Peer)), + ) { + Ok(updated) => { + // Only relay if we received new information. + if updated { + debug!("Address store entry for node {node} updated at {timestamp}"); + return Ok(relay); + } + } + Err(err) => { + // An error here is due to a fault in our address store. + error!("Error processing node announcement from {node}: {err}"); + } + } } } Ok(false) @@ -778,6 +815,7 @@ where ) -> Result<(), Error> { for proj_id in inventory { // TODO: Fire an event on routing update. + // FIXME: The timestamp we insert should be the announcement timestamp. if self .routing .insert(*proj_id, from, self.clock.timestamp())? diff --git a/radicle-node/src/test/peer.rs b/radicle-node/src/test/peer.rs index 0da3ede5..7ba223bb 100644 --- a/radicle-node/src/test/peer.rs +++ b/radicle-node/src/test/peer.rs @@ -4,9 +4,8 @@ use std::ops::{Deref, DerefMut}; use log::*; -use crate::address_book::{KnownAddress, Source}; +use crate::address; use crate::clock::{RefClock, Timestamp}; -use crate::collections::HashMap; use crate::git; use crate::git::Url; use crate::prelude::NodeId; @@ -21,8 +20,7 @@ use crate::test::simulator; use crate::{Link, LocalTime}; /// Service instantiation used for testing. -pub type Service = - service::Service, S, MockSigner>; +pub type Service = service::Service; #[derive(Debug)] pub struct Peer { @@ -82,7 +80,6 @@ where ..Config::default() }, ip, - vec![], storage, fastrand::Rng::new(), ) @@ -92,18 +89,14 @@ where name: &'static str, config: Config, ip: impl Into, - addrs: Vec<(net::SocketAddr, Source)>, storage: S, mut rng: fastrand::Rng, ) -> Self { - let addrs = addrs - .into_iter() - .map(|(addr, src)| (addr.ip(), KnownAddress::new(addr, src, None))) - .collect(); let local_time = LocalTime::now(); let clock = RefClock::from(local_time); let signer = MockSigner::new(&mut rng); let routing = routing::Table::memory().unwrap(); + let addrs = address::Book::memory().unwrap(); let service = Service::new(config, clock, routing, storage, addrs, signer, rng.clone()); let ip = ip.into(); let local_addr = net::SocketAddr::new(ip, rng.u16(..)); diff --git a/radicle-node/src/test/tests.rs b/radicle-node/src/test/tests.rs index 0a71bd73..61fd01aa 100644 --- a/radicle-node/src/test/tests.rs +++ b/radicle-node/src/test/tests.rs @@ -84,14 +84,7 @@ fn test_persistent_peer_connect() { connect: vec![bob.address(), eve.address()], ..Config::default() }; - let mut alice = Peer::config( - "alice", - config, - [7, 7, 7, 7], - vec![], - MockStorage::empty(), - rng, - ); + let mut alice = Peer::config("alice", config, [7, 7, 7, 7], MockStorage::empty(), rng); alice.initialize(); @@ -164,7 +157,6 @@ fn test_tracking() { ..Config::default() }, [7, 7, 7, 7], - vec![], MockStorage::empty(), fastrand::Rng::new(), ); @@ -387,7 +379,6 @@ fn test_persistent_peer_reconnect() { ..Config::default() }, [7, 7, 7, 7], - vec![], MockStorage::empty(), fastrand::Rng::new(), ); @@ -499,6 +490,7 @@ fn test_push_and_pull() { // Alice announces her refs. // We now expect Eve to fetch Alice's project from Alice. // Then we expect Bob to fetch Alice's project from Eve. + alice.clock().elapse(LocalDuration::from_secs(1)); // Make sure our announcement is fresh. alice.command(service::Command::AnnounceRefs(proj_id)); sim.run_while([&mut alice, &mut bob, &mut eve], |s| !s.is_settled()); diff --git a/radicle-node/src/transport.rs b/radicle-node/src/transport.rs index 473ef4ef..3fc4936e 100644 --- a/radicle-node/src/transport.rs +++ b/radicle-node/src/transport.rs @@ -5,7 +5,7 @@ use nakamoto::LocalTime; use nakamoto_net as nakamoto; use nakamoto_net::{Io, Link}; -use crate::address_book; +use crate::address; use crate::collections::HashMap; use crate::crypto; use crate::service::routing; @@ -37,7 +37,7 @@ impl nakamoto::Protocol for Transport where R: routing::Store, T: WriteStorage + 'static, - S: address_book::Store, + S: address::Store, G: crypto::Signer, { type Event = Event; diff --git a/radicle-node/src/wire.rs b/radicle-node/src/wire.rs index a1ef0995..9676fbca 100644 --- a/radicle-node/src/wire.rs +++ b/radicle-node/src/wire.rs @@ -11,7 +11,7 @@ use byteorder::{NetworkEndian, ReadBytesExt, WriteBytesExt}; use nakamoto_net as nakamoto; use nakamoto_net::Link; -use crate::address_book; +use crate::address; use crate::crypto::{PublicKey, Signature, Signer, Unverified}; use crate::decoder::Decoder; use crate::git; @@ -471,7 +471,7 @@ impl Wire { impl Wire where R: routing::Store, - S: address_book::Store, + S: address::Store, T: WriteStorage + 'static, G: Signer, { diff --git a/radicle-node/src/wire/message.rs b/radicle-node/src/wire/message.rs index b9efc213..6224b4fe 100644 --- a/radicle-node/src/wire/message.rs +++ b/radicle-node/src/wire/message.rs @@ -70,6 +70,17 @@ impl From for u8 { } } +impl From<&Address> for AddressType { + fn from(a: &Address) -> Self { + match a { + Address::Ipv4 { .. } => AddressType::Ipv4, + Address::Ipv6 { .. } => AddressType::Ipv6, + Address::Hostname { .. } => AddressType::Hostname, + Address::Onion { .. } => AddressType::Onion, + } + } +} + impl TryFrom for AddressType { type Error = u8; diff --git a/radicle/src/sql.rs b/radicle/src/sql.rs index 47e6d4ce..cb79206c 100644 --- a/radicle/src/sql.rs +++ b/radicle/src/sql.rs @@ -1,3 +1,4 @@ +use std::ops::Deref; use std::str::FromStr; use sqlite as sql; @@ -5,6 +6,7 @@ use sqlite::Value; use crate::crypto::PublicKey; use crate::identity::Id; +use crate::node; impl sql::ValueInto for Id { fn into(value: &Value) -> Option { @@ -35,3 +37,18 @@ impl sqlite::Bindable for &PublicKey { self.to_human().as_str().bind(stmt, i) } } + +impl sql::Bindable for node::Features { + fn bind(self, stmt: &mut sql::Statement<'_>, i: usize) -> sql::Result<()> { + (*self.deref() as i64).bind(stmt, i) + } +} + +impl sql::ValueInto for node::Features { + fn into(value: &Value) -> Option { + match value { + Value::Integer(bits) => Some(node::Features::from(*bits as u64)), + _ => None, + } + } +} diff --git a/radicle/src/storage/refs.rs b/radicle/src/storage/refs.rs index 01d398c0..35ebea89 100644 --- a/radicle/src/storage/refs.rs +++ b/radicle/src/storage/refs.rs @@ -268,7 +268,7 @@ impl SignedRefs { repo: &S, ) -> Result { let sigref = &*SIGNATURE_REF; - let parent = match repo.reference(remote, sigref).map_err(|e| dbg!(e)) { + let parent = match repo.reference(remote, sigref) { Ok(r) => Some(r.peel_to_commit()?), Err(git_ext::Error::Git(e)) if git_ext::is_not_found_err(&e) => None, Err(git_ext::Error::NotFound(_)) => None,