diff --git a/radicle-crypto/src/lib.rs b/radicle-crypto/src/lib.rs index e580e9f2..83e0fac1 100644 --- a/radicle-crypto/src/lib.rs +++ b/radicle-crypto/src/lib.rs @@ -442,6 +442,44 @@ impl sqlite::BindableWithIndex for &PublicKey { } } +#[cfg(feature = "sqlite")] +impl From<&Signature> for sqlite::Value { + fn from(sig: &Signature) -> Self { + sqlite::Value::Binary(sig.to_vec()) + } +} + +#[cfg(feature = "sqlite")] +impl TryFrom<&sqlite::Value> for Signature { + type Error = sqlite::Error; + + fn try_from(value: &sqlite::Value) -> Result { + match value { + sqlite::Value::Binary(s) => ed25519::Signature::from_slice(s) + .map_err(|e| sqlite::Error { + code: None, + message: Some(e.to_string()), + }) + .map(Self), + _ => Err(sqlite::Error { + code: None, + message: Some("sql: invalid column type for signature".to_owned()), + }), + } + } +} + +#[cfg(feature = "sqlite")] +impl sqlite::BindableWithIndex for &Signature { + fn bind( + self, + stmt: &mut sqlite::Statement<'_>, + i: I, + ) -> sqlite::Result<()> { + sqlite::Value::from(self).bind(stmt, i) + } +} + pub mod keypair { use super::*; diff --git a/radicle-node/src/runtime.rs b/radicle-node/src/runtime.rs index 01044908..1011846c 100644 --- a/radicle-node/src/runtime.rs +++ b/radicle-node/src/runtime.rs @@ -27,7 +27,7 @@ use crate::control; use crate::crypto::Signer; use crate::node::{routing, NodeId}; use crate::service::message::NodeAnnouncement; -use crate::service::{tracking, Event}; +use crate::service::{gossip, tracking, Event}; use crate::wire::Wire; use crate::wire::{self, Decode}; use crate::worker; @@ -48,6 +48,9 @@ pub enum Error { /// A tracking database error. #[error("tracking database error: {0}")] Tracking(#[from] tracking::Error), + /// A gossip database error. + #[error("gossip database error: {0}")] + Gossip(#[from] gossip::Error), /// An I/O error. #[error("i/o error: {0}")] Io(#[from] io::Error), @@ -141,7 +144,10 @@ impl Runtime { let policy = config.policy; log::info!(target: "node", "Opening address book {}..", address_db.display()); - let mut addresses = address::Book::open(address_db)?; + let mut addresses = address::Book::open(address_db.clone())?; + + log::info!(target: "node", "Opening gossip store from {}..", address_db.display()); + let gossip = gossip::Store::open(address_db)?; // Nb. same database as address book. log::info!(target: "node", "Opening routing table {}..", routing_db.display()); let routing = routing::Table::open(routing_db)?; @@ -204,6 +210,7 @@ impl Runtime { routing, storage.clone(), addresses, + gossip, tracking, signer.clone(), rng, diff --git a/radicle-node/src/service.rs b/radicle-node/src/service.rs index 43b40218..e80a105f 100644 --- a/radicle-node/src/service.rs +++ b/radicle-node/src/service.rs @@ -1,14 +1,16 @@ #![allow(clippy::too_many_arguments)] #![allow(clippy::collapsible_match)] #![allow(clippy::collapsible_if)] +#![warn(clippy::unwrap_used)] pub mod filter; +pub mod gossip; pub mod io; pub mod limitter; pub mod message; pub mod session; use std::collections::hash_map::Entry; -use std::collections::{BTreeMap, HashMap, HashSet}; +use std::collections::{HashMap, HashSet}; use std::ops::{Deref, DerefMut}; use std::sync::Arc; use std::{fmt, time}; @@ -50,7 +52,6 @@ pub use crate::service::session::Session; pub use radicle::node::tracking::config as tracking; -use self::gossip::Gossip; use self::io::Outbox; use self::limitter::RateLimiter; use self::message::InventoryAnnouncement; @@ -203,7 +204,7 @@ pub struct Service { /// Tracking policy configuration. tracking: tracking::Config, /// State relating to gossip. - gossip: Gossip, + gossip: gossip::Store, /// Peer sessions, currently or recently connected. sessions: Sessions, /// Clock. Tells the time. @@ -262,6 +263,7 @@ where routing: R, storage: S, addresses: A, + gossip: gossip::Store, tracking: tracking::Config, signer: G, rng: Rng, @@ -280,7 +282,7 @@ where node, clock, routing, - gossip: Gossip::default(), + gossip, outbox: Outbox::default(), limiter: RateLimiter::default(), sessions, @@ -483,7 +485,7 @@ where .inventory() .and_then(|i| self.announce_inventory(i)) { - error!(target: "service", "Error announcing inventory: {}", err); + error!(target: "service", "Error announcing inventory: {err}"); } self.outbox.wakeup(ANNOUNCE_INTERVAL); self.last_announce = now; @@ -492,8 +494,15 @@ where trace!(target: "service", "Running 'prune' task..."); if let Err(err) = self.prune_routing_entries(&now) { - error!("Error pruning routing entries: {}", err); + error!(target: "service", "Error pruning routing entries: {err}"); } + if let Err(err) = self + .gossip + .prune((now - self.config.limits.gossip_max_age).as_millis()) + { + error!(target: "service", "Error pruning gossip entries: {err}"); + } + self.outbox.wakeup(PRUNE_INTERVAL); self.last_prune = now; } @@ -886,26 +895,36 @@ where let now = self.clock; let timestamp = message.timestamp(); let relay = self.config.relay; - let peer = self - .gossip - .nodes - .entry(*announcer) - .or_insert_with(Node::default); // Don't allow messages from too far in the future. if timestamp.saturating_sub(now.as_millis()) > MAX_TIME_DELTA.as_millis() as u64 { return Err(session::Error::InvalidTimestamp(timestamp)); } - - match message { - AnnouncementMessage::Inventory(message) => { - // Discard inventory messages we've already seen, otherwise update - // out last seen time. - if !peer.inventory_announced(announcement.clone()) { + // Check ref signatures validity. + if let AnnouncementMessage::Refs(message) = message { + for theirs in message.refs.iter() { + if theirs.verify(&theirs.id).is_err() { + warn!(target: "service", "Peer {relayer} relayed refs announcement with invalid signature for {}", theirs.id); + return Err(session::Error::Misbehavior); + } + } + } + // Discard announcement messages we've already seen, otherwise update out last seen time. + match self.gossip.announced(announcer, announcement) { + Ok(fresh) => { + if !fresh { trace!(target: "service", "Ignoring stale inventory announcement from {announcer} (t={})", self.time()); return Ok(false); } + } + Err(e) => { + error!(target: "service", "Error updating gossip entry from {announcer}: {e}"); + return Ok(false); + } + } + match message { + AnnouncementMessage::Inventory(message) => { match self.sync_routing(&message.inventory, *announcer, message.timestamp) { Ok(synced) => { if synced.is_empty() { @@ -914,7 +933,7 @@ where } } Err(e) => { - error!(target: "service", "Error processing inventory from {}: {}", announcer, e); + error!(target: "service", "Error processing inventory from {announcer}: {e}"); return Ok(false); } } @@ -957,13 +976,6 @@ where } // Process a peer inventory update announcement by (maybe) fetching. AnnouncementMessage::Refs(message) => { - for theirs in message.refs.iter() { - if theirs.verify(&theirs.id).is_err() { - warn!(target: "service", "Peer {relayer} relayed refs announcement with invalid signature for {}", theirs.id); - return Err(session::Error::Misbehavior); - } - } - // We update inventories when receiving ref announcements, as these could come // from a new repository being initialized. if let Ok(result) = @@ -978,12 +990,6 @@ where info!(target: "service", "Routing table updated for {} with seed {announcer}", message.rid); } } - // Discard announcement messages we've already seen, otherwise update - // our last seen time. - if !peer.refs_announced(message.rid, announcement.clone()) { - trace!(target: "service", "Ignoring stale refs announcement from {announcer} (time={timestamp})"); - return Ok(false); - } // Check if the announcer is in sync with our own refs, and if so emit an event. // This event is used for showing sync progress to users. @@ -1042,13 +1048,6 @@ where .. }, ) => { - // Discard node messages we've already seen, otherwise update - // our last seen time. - if !peer.node_announced(announcement.clone()) { - trace!(target: "service", "Ignoring stale node announcement from {announcer}"); - return Ok(false); - } - // 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) { @@ -1174,14 +1173,30 @@ where } } (session::State::Connected { .. }, Message::Subscribe(subscribe)) => { - for ann in self + // Filter announcements by interest. + match self .gossip - // Filter announcements by interest. .filtered(&subscribe.filter, subscribe.since, subscribe.until) - // Don't send announcements authored by the remote, back to the remote. - .filter(|ann| &ann.node != remote) { - self.outbox.write(peer, ann.into()); + Ok(anns) => { + for ann in anns { + let ann = match ann { + Ok(a) => a, + Err(e) => { + error!(target: "service", "Error reading gossip message from store: {e}"); + continue; + } + }; + // Don't send announcements authored by the remote, back to the remote. + if ann.node == *remote { + continue; + } + self.outbox.write(peer, ann.into()); + } + } + Err(e) => { + error!(target: "service", "Error querying gossip messages from store: {e}"); + } } peer.subscribe = Some(subscribe); } @@ -1720,75 +1735,6 @@ pub enum LookupError { Repository(#[from] RepositoryError), } -/// Keeps track of the most recent announcements of a node. -#[derive(Default, Debug)] -pub struct Node { - /// Last ref announcements (per project). - pub last_refs: HashMap, - /// Last inventory announcement. - pub last_inventory: Option, - /// Last node announcement. - pub last_node: Option, -} - -impl Node { - /// Process a refs announcement for the given node. - /// Returns `true` if the timestamp was updated. - pub fn refs_announced(&mut self, id: Id, ann: Announcement) -> bool { - match self.last_refs.entry(id) { - Entry::Vacant(e) => { - e.insert(ann); - return true; - } - Entry::Occupied(mut e) => { - let last = e.get_mut(); - - if ann.timestamp() > last.timestamp() { - *last = ann; - return true; - } - } - } - false - } - - /// Process an inventory announcement for the given node. - /// Returns `true` if the timestamp was updated. - pub fn inventory_announced(&mut self, ann: Announcement) -> bool { - match &mut self.last_inventory { - Some(last) => { - if ann.timestamp() > last.timestamp() { - *last = ann; - return true; - } - } - None => { - self.last_inventory = Some(ann); - return true; - } - } - false - } - - /// Process a node announcement for the given node. - /// Returns `true` if the timestamp was updated. - pub fn node_announced(&mut self, ann: Announcement) -> bool { - match &mut self.last_node { - Some(last) => { - if ann.timestamp() > last.timestamp() { - *last = ann; - return true; - } - } - None => { - self.last_node = Some(ann); - return true; - } - } - false - } -} - #[derive(Debug, Clone)] /// Holds currently (or recently) connected peers. pub struct Sessions(AddressBook); @@ -1842,100 +1788,3 @@ impl DerefMut for Sessions { &mut self.0 } } - -pub mod gossip { - use super::*; - use crate::service::filter::Filter; - - #[derive(Default, Debug)] - pub struct Gossip { - // FIXME: This should be loaded from the address store. - /// Keeps track of node announcements. - pub nodes: BTreeMap, - } - - impl Gossip { - pub fn filtered<'a>( - &'a self, - filter: &'a Filter, - start: Timestamp, - end: Timestamp, - ) -> impl Iterator + '_ { - self.nodes - .values() - .flat_map(|n| { - [&n.last_node, &n.last_inventory] - .into_iter() - .flatten() - .chain(n.last_refs.values()) - .cloned() - .collect::>() - }) - .filter(move |ann| ann.timestamp() >= start && ann.timestamp() < end) - .filter(move |ann| ann.matches(filter)) - } - } - - pub fn handshake( - node: NodeAnnouncement, - now: Timestamp, - storage: &S, - signer: &G, - filter: Filter, - ) -> Vec { - let inventory = match storage.inventory() { - Ok(i) => i, - Err(e) => { - error!("Error getting local inventory for handshake: {}", e); - // Other than crashing the node completely, there's nothing we can do - // here besides returning an empty inventory and logging an error. - vec![] - } - }; - - vec![ - Message::node(node, signer), - Message::inventory(gossip::inventory(now, inventory), signer), - Message::subscribe( - filter, - now - SUBSCRIBE_BACKLOG_DELTA.as_millis() as u64, - Timestamp::MAX, - ), - ] - } - - pub fn node(config: &Config, timestamp: Timestamp) -> NodeAnnouncement { - let features = config.features(); - let alias = config.alias.clone(); - let addresses: BoundedVec<_, ADDRESS_LIMIT> = config - .external_addresses - .clone() - .try_into() - .expect("external addresses are within the limit"); - - NodeAnnouncement { - features, - timestamp, - alias, - addresses, - nonce: 0, - } - } - - pub fn inventory(timestamp: Timestamp, inventory: Vec) -> InventoryAnnouncement { - type Inventory = BoundedVec; - - if inventory.len() > Inventory::max() { - error!( - target: "service", - "inventory announcement limit ({}) exceeded, other nodes will see only some of your projects", - inventory.len() - ); - } - - InventoryAnnouncement { - inventory: BoundedVec::truncate(inventory), - timestamp, - } - } -} diff --git a/radicle-node/src/service/gossip.rs b/radicle-node/src/service/gossip.rs new file mode 100644 index 00000000..3a14cd4e --- /dev/null +++ b/radicle-node/src/service/gossip.rs @@ -0,0 +1,70 @@ +pub mod store; + +use super::*; +use crate::service::filter::Filter; + +pub use store::Error; +pub use store::GossipStore as Store; + +pub fn handshake( + node: NodeAnnouncement, + now: Timestamp, + storage: &S, + signer: &G, + filter: Filter, +) -> Vec { + let inventory = match storage.inventory() { + Ok(i) => i, + Err(e) => { + error!("Error getting local inventory for handshake: {}", e); + // Other than crashing the node completely, there's nothing we can do + // here besides returning an empty inventory and logging an error. + vec![] + } + }; + + vec![ + Message::node(node, signer), + Message::inventory(gossip::inventory(now, inventory), signer), + Message::subscribe( + filter, + now - SUBSCRIBE_BACKLOG_DELTA.as_millis() as u64, + Timestamp::MAX, + ), + ] +} + +pub fn node(config: &Config, timestamp: Timestamp) -> NodeAnnouncement { + let features = config.features(); + let alias = config.alias.clone(); + let addresses: BoundedVec<_, ADDRESS_LIMIT> = config + .external_addresses + .clone() + .try_into() + .expect("external addresses are within the limit"); + + NodeAnnouncement { + features, + timestamp, + alias, + addresses, + nonce: 0, + } +} + +pub fn inventory(timestamp: Timestamp, inventory: Vec) -> InventoryAnnouncement { + type Inventory = BoundedVec; + + if inventory.len() > Inventory::max() { + error!( + target: "service", + "inventory announcement limit ({}) exceeded, other nodes will see only some of your projects", + inventory.len() + ); + } + + InventoryAnnouncement { + inventory: BoundedVec::truncate(inventory), + timestamp, + } +} diff --git a/radicle-node/src/service/gossip/store.rs b/radicle-node/src/service/gossip/store.rs new file mode 100644 index 00000000..8d12a553 --- /dev/null +++ b/radicle-node/src/service/gossip/store.rs @@ -0,0 +1,279 @@ +use std::{fmt, io, path::Path}; + +use radicle::crypto::Signature; +use sqlite as sql; +use thiserror::Error; + +use crate::node::NodeId; +use crate::prelude::{Filter, Timestamp}; +use crate::service::message::{ + Announcement, AnnouncementMessage, InventoryAnnouncement, NodeAnnouncement, RefsAnnouncement, +}; +use crate::wire; +use crate::wire::Decode; + +#[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), +} + +/// Keeps track of the latest received gossip messages for each node. +/// Grows linearly with the number of nodes on the network. +pub struct GossipStore { + db: sql::Connection, +} + +impl fmt::Debug for GossipStore { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("GossipStore").finish() + } +} + +impl GossipStore { + /// Open a gossip 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_with_flags( + path, + sqlite::OpenFlags::new().set_read_write().set_full_mutex(), + )?; + + Ok(Self { db }) + } + + /// Prune announcements older than the cutoff time. + pub fn prune(&mut self, cutoff: Timestamp) -> Result { + let mut stmt = self + .db + .prepare("DELETE FROM `announcements` WHERE timestamp < ?1")?; + + stmt.bind((1, cutoff.try_into().unwrap_or(i64::MAX)))?; + stmt.next()?; + + Ok(self.db.change_count()) + } + + /// Process an announcement for the given node. + /// Returns `true` if the timestamp was updated or the announcement wasn't there before. + pub fn announced(&mut self, nid: &NodeId, ann: &Announcement) -> Result { + let mut stmt = self.db.prepare( + "INSERT INTO `announcements` (node, repo, type, message, signature, timestamp) + VALUES (?1, ?2, ?3, ?4, ?5, ?6) + ON CONFLICT DO UPDATE + SET message = ?4, signature = ?5, timestamp = ?6 + WHERE timestamp < ?6", + )?; + stmt.bind((1, nid))?; + + match &ann.message { + AnnouncementMessage::Node(msg) => { + stmt.bind((2, sql::Value::String(String::new())))?; + stmt.bind((3, &GossipType::Node))?; + stmt.bind((4, msg))?; + } + AnnouncementMessage::Refs(msg) => { + stmt.bind((2, &msg.rid))?; + stmt.bind((3, &GossipType::Refs))?; + stmt.bind((4, msg))?; + } + AnnouncementMessage::Inventory(msg) => { + stmt.bind((2, sql::Value::String(String::new())))?; + stmt.bind((3, &GossipType::Inventory))?; + stmt.bind((4, msg))?; + } + } + stmt.bind((5, &ann.signature))?; + stmt.bind((6, ann.message.timestamp().try_into().unwrap_or(i64::MAX)))?; + stmt.next()?; + + Ok(self.db.change_count() > 0) + } + + /// Get all the latest gossip messages of all nodes, filtered by inventory filter and + /// announcement timestamps. + /// + /// # Panics + /// + /// Panics if `from` > `to`. + /// + pub fn filtered<'a>( + &'a self, + filter: &'a Filter, + from: Timestamp, + to: Timestamp, + ) -> Result> + 'a, Error> { + let mut stmt = self.db.prepare( + "SELECT node, type, message, signature, timestamp + FROM announcements + WHERE timestamp >= ?1 and timestamp < ?2 + ORDER BY timestamp", + )?; + assert!(from <= to); + + stmt.bind((1, i64::try_from(from).unwrap_or(i64::MAX)))?; + stmt.bind((2, i64::try_from(to).unwrap_or(i64::MAX)))?; + + Ok(stmt + .into_iter() + .map(|row| { + let row = row?; + let node = row.read::("node"); + let gt = row.read::("type"); + let message = match gt { + GossipType::Refs => { + let ann = row.read::("message"); + AnnouncementMessage::Refs(ann) + } + GossipType::Inventory => { + let ann = row.read::("message"); + AnnouncementMessage::Inventory(ann) + } + GossipType::Node => { + let ann = row.read::("message"); + AnnouncementMessage::Node(ann) + } + }; + let signature = row.read::("signature"); + let timestamp = row.read::("timestamp"); + + debug_assert_eq!(timestamp, message.timestamp() as i64); + + Ok(Announcement { + node, + message, + signature, + }) + }) + .filter(|ann| match ann { + Ok(a) => a.matches(filter), + Err(_) => true, + })) + } +} + +impl TryFrom<&sql::Value> for NodeAnnouncement { + type Error = sql::Error; + + fn try_from(value: &sql::Value) -> Result { + match value { + sql::Value::Binary(bytes) => { + let mut reader = io::Cursor::new(bytes); + NodeAnnouncement::decode(&mut reader).map_err(wire::Error::into) + } + _ => Err(sql::Error { + code: None, + message: Some("sql: invalid type for node announcement".to_owned()), + }), + } + } +} + +impl sql::BindableWithIndex for &NodeAnnouncement { + fn bind(self, stmt: &mut sql::Statement<'_>, i: I) -> sql::Result<()> { + wire::serialize(self).bind(stmt, i) + } +} + +impl TryFrom<&sql::Value> for RefsAnnouncement { + type Error = sql::Error; + + fn try_from(value: &sql::Value) -> Result { + match value { + sql::Value::Binary(bytes) => { + let mut reader = io::Cursor::new(bytes); + RefsAnnouncement::decode(&mut reader).map_err(wire::Error::into) + } + _ => Err(sql::Error { + code: None, + message: Some("sql: invalid type for refs announcement".to_owned()), + }), + } + } +} + +impl sql::BindableWithIndex for &RefsAnnouncement { + fn bind(self, stmt: &mut sql::Statement<'_>, i: I) -> sql::Result<()> { + wire::serialize(self).bind(stmt, i) + } +} + +impl TryFrom<&sql::Value> for InventoryAnnouncement { + type Error = sql::Error; + + fn try_from(value: &sql::Value) -> Result { + match value { + sql::Value::Binary(bytes) => { + let mut reader = io::Cursor::new(bytes); + InventoryAnnouncement::decode(&mut reader).map_err(wire::Error::into) + } + _ => Err(sql::Error { + code: None, + message: Some("sql: invalid type for inventory announcement".to_owned()), + }), + } + } +} + +impl sql::BindableWithIndex for &InventoryAnnouncement { + fn bind(self, stmt: &mut sql::Statement<'_>, i: I) -> sql::Result<()> { + wire::serialize(self).bind(stmt, i) + } +} + +impl From for sql::Error { + fn from(other: wire::Error) -> Self { + sql::Error { + code: None, + message: Some(other.to_string()), + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum GossipType { + Refs, + Node, + Inventory, +} + +impl fmt::Display for GossipType { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Refs => write!(f, "refs"), + Self::Node => write!(f, "node"), + Self::Inventory => write!(f, "inventory"), + } + } +} + +impl sql::BindableWithIndex for &GossipType { + fn bind(self, stmt: &mut sql::Statement<'_>, i: I) -> sql::Result<()> { + self.to_string().as_str().bind(stmt, i) + } +} + +impl TryFrom<&sql::Value> for GossipType { + type Error = sql::Error; + + fn try_from(value: &sql::Value) -> Result { + match value { + sql::Value::String(s) => match s.as_str() { + "refs" => Ok(Self::Refs), + "node" => Ok(Self::Node), + "inventory" => Ok(Self::Inventory), + other => Err(sql::Error { + code: None, + message: Some(format!("unknown gossip type '{other}'")), + }), + }, + _ => Err(sql::Error { + code: None, + message: Some("sql: invalid type for gossip type".to_owned()), + }), + } + } +} diff --git a/radicle-node/src/service/message.rs b/radicle-node/src/service/message.rs index 6f621bbe..7d1611d9 100644 --- a/radicle-node/src/service/message.rs +++ b/radicle-node/src/service/message.rs @@ -525,6 +525,7 @@ impl ZeroBytes { } #[cfg(test)] +#[allow(clippy::unwrap_used)] mod tests { use super::*; use crate::prelude::*; diff --git a/radicle-node/src/test/peer.rs b/radicle-node/src/test/peer.rs index 0db729a9..0f8fe09f 100644 --- a/radicle-node/src/test/peer.rs +++ b/radicle-node/src/test/peer.rs @@ -99,26 +99,33 @@ where pub struct Config { pub config: service::Config, pub addrs: address::Book, + pub gossip: gossip::Store, pub local_time: LocalTime, pub policy: Policy, pub scope: Scope, pub signer: G, pub rng: fastrand::Rng, + pub tmp: tempfile::TempDir, } impl Default for Config { fn default() -> Self { let mut rng = fastrand::Rng::new(); let signer = MockSigner::new(&mut rng); + let tmp = tempfile::TempDir::new().unwrap(); + let addrs = address::Book::open(tmp.path().join("addresses.db")).unwrap(); + let gossip = gossip::Store::open(tmp.path().join("addresses.db")).unwrap(); Config { config: service::Config::test(Alias::from_str("mocky").unwrap()), - addrs: address::Book::memory().unwrap(), + addrs, + gossip, local_time: LocalTime::now(), policy: Policy::default(), scope: Scope::default(), signer, rng, + tmp, } } } @@ -157,7 +164,6 @@ where let routing = routing::Table::memory().unwrap(); let tracking = tracking::Store::::memory().unwrap(); let mut tracking = tracking::Config::new(config.policy, config.scope, tracking); - let tempdir = tempfile::tempdir().unwrap(); let id = *config.signer.public_key(); let ip = ip.into(); let local_addr = net::SocketAddr::new(ip, config.rng.u16(..)); @@ -176,6 +182,7 @@ where routing, storage, config.addrs, + config.gossip, tracking, config.signer, config.rng.clone(), @@ -191,7 +198,7 @@ where local_addr, rng: config.rng, initialized: false, - tempdir, + tempdir: config.tmp, } } diff --git a/radicle/src/node/address/schema.sql b/radicle/src/node/address/schema.sql index e9d1a352..3abd3c70 100644 --- a/radicle/src/node/address/schema.sql +++ b/radicle/src/node/address/schema.sql @@ -37,3 +37,26 @@ create table if not exists "addresses" ( unique ("node", "type", "value") -- ) strict; + +create table if not exists "announcements" ( + -- Node ID. + "node" text not null references "nodes" ("id"), + -- Repo ID, if any, for example in ref announcements. + "repo" text not null, + -- Announcement type. + -- + -- Valid values are: + -- + -- "refs" + -- "node" + -- "inventory" + "type" text not null, + -- Announcement message in wire format (binary). + "message" blob not null, + -- Signature over message. + "signature" blob not null, + -- Announcement timestamp. + "timestamp" integer not null, + -- + unique ("node", "repo", "type") +) strict; diff --git a/radicle/src/node/address/store.rs b/radicle/src/node/address/store.rs index 9307662a..b9d82e1d 100644 --- a/radicle/src/node/address/store.rs +++ b/radicle/src/node/address/store.rs @@ -38,13 +38,25 @@ impl fmt::Debug for Book { } } +impl From for Book { + fn from(db: sql::Connection) -> Self { + Self { db } + } +} + 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)?; + let db = sql::Connection::open_with_flags( + path, + sqlite::OpenFlags::new() + .set_create() + .set_read_write() + .set_full_mutex(), + )?; db.execute(Self::SCHEMA)?; Ok(Self { db }) diff --git a/radicle/src/node/config.rs b/radicle/src/node/config.rs index 0b4da321..08a2ee83 100644 --- a/radicle/src/node/config.rs +++ b/radicle/src/node/config.rs @@ -56,6 +56,9 @@ pub struct Limits { /// How long to keep a routing table entry before being pruned. #[serde(with = "crate::serde_ext::localtime::duration")] pub routing_max_age: LocalDuration, + /// How long to keep a gossip message entry before pruning it. + #[serde(with = "crate::serde_ext::localtime::duration")] + pub gossip_max_age: LocalDuration, /// Maximum number of concurrent fetches per per connection. pub fetch_concurrency: usize, /// Rate limitter settings. @@ -67,7 +70,8 @@ impl Default for Limits { fn default() -> Self { Self { routing_max_size: 1000, - routing_max_age: LocalDuration::from_mins(7 * 24 * 60), + routing_max_age: LocalDuration::from_mins(7 * 24 * 60), // One week + gossip_max_age: LocalDuration::from_mins(2 * 7 * 24 * 60), // Two weeks fetch_concurrency: 1, rate: RateLimits::default(), }