From f9df360171b64ffa7e3cbe96a3145330dabc3d86 Mon Sep 17 00:00:00 2001 From: cloudhead Date: Tue, 7 Nov 2023 16:27:57 +0100 Subject: [PATCH] node: Persist historical gossip messages Instead of storing received gossip messages in an in-memory `BTreeMap`, we persist them to the database (`addresses.db`). There are two reasons for this: 1. Node restarts do not wipe the state, meaning that it is much less likely that some gossip message will get lost before reaching all nodes. 2. This will allow us in the next commit to know how far behind we are after a restart, to be able to request the correct time range of gossip messages that we missed while offline. Note that we now share the addresses.db file between two connections and stores, which we don't do anywhere else. Having multiple connections to the same database is fine, but the way we're having to do it is a bit awkward. In a future patch, we could pull out the database schema from the `address` module and rename the database to make things clearer. --- radicle-crypto/src/lib.rs | 38 +++ radicle-node/src/runtime.rs | 11 +- radicle-node/src/service.rs | 267 +++++----------------- radicle-node/src/service/gossip.rs | 70 ++++++ radicle-node/src/service/gossip/store.rs | 279 +++++++++++++++++++++++ radicle-node/src/service/message.rs | 1 + radicle-node/src/test/peer.rs | 13 +- radicle/src/node/address/schema.sql | 23 ++ radicle/src/node/address/store.rs | 14 +- radicle/src/node/config.rs | 6 +- 10 files changed, 506 insertions(+), 216 deletions(-) create mode 100644 radicle-node/src/service/gossip.rs create mode 100644 radicle-node/src/service/gossip/store.rs 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(), }