mab122
/
radicle-heartwood-lfs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

node: Handle node announcements 

To build an address book, peers communicate via gossip, sending
`NodeAnnouncement` messages.

These need to be handled by storing the addresses in an address book
and the addresses need to be persisted to disk.

Node announcement messages should also be relayed to peers like other
gossip messages.
This commit is contained in:
Alexis Sellier 2022-10-11 15:51:12 +02:00
parent 0482f0ce04
commit 07add86bb6
No known key found for this signature in database
15 changed files with 784 additions and 524 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
radicle-node/src/address_manager.rs → radicle-node/src/address.rs
View File

@ -1,4 +1,8 @@
use crate::address_book::Store;
mod store;
mod types;
pub use store::*;
pub use types::*;
#[derive(Debug)]
pub struct AddressManager<S> {

37
radicle-node/src/address/schema.sql Normal file
View File

@ -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;

497
radicle-node/src/address/store.rs Normal file
View File

@ -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<P: AsRef<Path>>(path: P) -> Result<Self, Error> {
let db = sql::Connection::open(path)?;
db.execute(Self::SCHEMA)?;
Ok(Self { db })
}
/// Create a new in-memory address book.
pub fn memory() -> Result<Self, Error> {
let db = sql::Connection::open(":memory:")?;
db.execute(Self::SCHEMA)?;
Ok(Self { db })
}
}
impl Store for Book {
fn get(&self, node: &NodeId) -> Result<Option<types::Node>, 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::<node::Features, _>(0);
let alias = row.get::<String, _>(1);
let timestamp = row.get::<i64, _>(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::<AddressType, _>(0);
let addr = row.get::<Address, _>(1);
let source = row.get::<Source, _>(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<usize, Error> {
let row = self
.db
.prepare("SELECT COUNT(*) FROM addresses")?
.into_cursor()
.next()
.unwrap()
.unwrap();
let count = row.get::<i64, _>(0) as usize;
Ok(count)
}
fn insert(
&mut self,
node: &NodeId,
features: node::Features,
alias: &str,
timestamp: Timestamp,
addrs: impl IntoIterator<Item = KnownAddress>,
) -> Result<bool, Error> {
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<bool, Error> {
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<Box<dyn Iterator<Item = (NodeId, KnownAddress)>>, 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::<AddressType, _>(1);
let addr = row.get::<Address, _>(2);
let source = row.get::<Source, _>(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<Option<types::Node>, 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<Item = KnownAddress>,
) -> Result<bool, Error>;
/// Remove an address from the store.
fn remove(&mut self, id: &NodeId) -> Result<bool, Error>;
/// Returns the number of addresses.
fn len(&self) -> Result<usize, Error>;
/// Returns true if there are no addresses.
fn is_empty(&self) -> Result<bool, Error> {
self.len().map(|l| l == 0)
}
/// Get the address entries in the store.
fn entries(&self) -> Result<Box<dyn Iterator<Item = (NodeId, KnownAddress)>>, Error>;
}
impl sql::ValueInto for Address {
fn into(value: &sql::Value) -> Option<Self> {
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<Self> {
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<Self> {
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::<NodeId>(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::<NodeId>(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::<NodeId>(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::<NodeId>(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::<NodeId>(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::<NodeId>(1);
let bob = arbitrary::gen::<NodeId>(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::<NodeId>(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::<Vec<_>>();
actual.sort_by_key(|(i, _)| *i);
expected.sort_by_key(|(i, _)| *i);
assert_eq!(cache.len().unwrap(), actual.len());
assert_eq!(actual, expected);
}
}

135
radicle-node/src/address/types.rs Normal file
View File

@ -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<K, V> {
inner: HashMap<K, V>,
rng: fastrand::Rng,
}
impl<K, V> AddressBook<K, V> {
/// 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<Item = &K> {
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::<Vec<_>>();
self.rng.shuffle(&mut keys);
keys.into_iter()
}
}
impl<K, V> Deref for AddressBook<K, V> {
type Target = HashMap<K, V>;
fn deref(&self) -> &Self::Target {
&self.inner
}
}
impl<K, V> DerefMut for AddressBook<K, V> {
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<KnownAddress>,
/// 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<LocalTime>,
/// Last time this address was tried.
pub last_attempt: Option<LocalTime>,
}
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"),
}
}
}

468
radicle-node/src/address_book.rs
View File

@ -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<K, V> {
inner: HashMap<K, V>,
rng: fastrand::Rng,
}
impl<K, V> AddressBook<K, V> {
/// 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<Item = &K> {
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::<Vec<_>>();
self.rng.shuffle(&mut keys);
keys.into_iter()
}
}
impl<K, V> Deref for AddressBook<K, V> {
type Target = HashMap<K, V>;
fn deref(&self) -> &Self::Target {
&self.inner
}
}
impl<K, V> DerefMut for AddressBook<K, V> {
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<LocalTime>,
/// Last time this address was sampled.
#[serde(with = "local_time")]
pub last_sampled: Option<LocalTime>,
/// Last time this address was tried.
#[serde(with = "local_time")]
pub last_attempt: Option<LocalTime>,
/// Last time this peer was seen alive.
#[serde(with = "local_time")]
pub last_active: Option<LocalTime>,
}
impl KnownAddress {
/// Create a new known address.
pub fn new(addr: net::SocketAddr, source: Source, last_active: Option<LocalTime>) -> 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<net::IpAddr, KnownAddress>,
file: fs::File,
}
impl Cache {
/// Open an existing cache.
pub fn open<P: AsRef<Path>>(path: P) -> io::Result<Self> {
fs::OpenOptions::new()
.read(true)
.write(true)
.open(path)
.and_then(Self::from)
}
/// Create a new cache.
pub fn create<P: AsRef<Path>>(path: P) -> io::Result<Self> {
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<Self> {
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<KnownAddress> {
self.addrs.remove(ip)
}
fn insert(&mut self, ip: net::IpAddr, ka: KnownAddress) -> bool {
<std::collections::HashMap<_, _> as Store>::insert(&mut self.addrs, ip, ka)
}
fn iter<'a>(&'a self) -> Box<dyn Iterator<Item = (&net::IpAddr, &KnownAddress)> + '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<KnownAddress>;
/// Return an iterator over the known addresses.
fn iter<'a>(&'a self) -> Box<dyn Iterator<Item = (&net::IpAddr, &KnownAddress)> + '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<S: net::ToSocketAddrs>(
&mut self,
seeds: impl Iterator<Item = S>,
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<net::IpAddr, KnownAddress> {
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<KnownAddress> {
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<dyn Iterator<Item = (&net::IpAddr, &KnownAddress)> + '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<net::IpAddr, KnownAddress> {
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<KnownAddress> {
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<dyn Iterator<Item = (&net::IpAddr, &KnownAddress)> + '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<Option<LocalTime>, D::Error>
where
D: Deserializer<'de>,
{
let value: Option<u64> = Deserialize::deserialize(deserializer)?;
if let Some(value) = value {
Ok(Some(LocalTime::from_secs(value)))
} else {
Ok(None)
}
}
pub fn serialize<S>(value: &Option<LocalTime>, serializer: S) -> Result<S::Ok, S::Error>
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::<Vec<_>>();
actual.sort_by_key(|(i, _)| *i);
expected.sort_by_key(|(i, _)| *i);
assert_eq!(actual, expected);
}
}
}

11
radicle-node/src/client.rs
View File

@ -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<R: Reactor> Client<R> {
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);

3
radicle-node/src/lib.rs
View File

@ -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;

88
radicle-node/src/service.rs
View File

@ -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<R, A, S, G> {
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<R, A, S, G> {
clock: RefClock,
/// Interface to the I/O reactor.
reactor: Reactor,
/// Peer address manager.
addrmgr: AddressManager<A>,
/// 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<R, A, S, G> Service<R, A, S, G>
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())?

13
radicle-node/src/test/peer.rs
View File

@ -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<S> =
service::Service<routing::Table, HashMap<net::IpAddr, KnownAddress>, S, MockSigner>;
pub type Service<S> = service::Service<routing::Table, address::Book, S, MockSigner>;
#[derive(Debug)]
pub struct Peer<S> {
@ -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<net::IpAddr>,
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(..));

12
radicle-node/src/test/tests.rs
View File

@ -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());

4
radicle-node/src/transport.rs
View File

@ -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<R, S, T, G> nakamoto::Protocol for Transport<R, S, T, G>
where
R: routing::Store,
T: WriteStorage + 'static,
S: address_book::Store,
S: address::Store,
G: crypto::Signer,
{
type Event = Event;

4
radicle-node/src/wire.rs
View File

@ -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<R, S, T, G> Wire<R, S, T, G> {
impl<R, S, T, G> Wire<R, S, T, G>
where
R: routing::Store,
S: address_book::Store,
S: address::Store,
T: WriteStorage + 'static,
G: Signer,
{

11
radicle-node/src/wire/message.rs
View File

@ -70,6 +70,17 @@ impl From<AddressType> 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<u8> for AddressType {
type Error = u8;

17
radicle/src/sql.rs
View File

@ -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<Self> {
@ -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<Self> {
match value {
Value::Integer(bits) => Some(node::Features::from(*bits as u64)),
_ => None,
}
}
}

2
radicle/src/storage/refs.rs
View File

@ -268,7 +268,7 @@ impl SignedRefs<Verified> {
repo: &S,
) -> Result<Updated, Error> {
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,