From 5472931ff0ba947f62b4dd85bab7a32b8cc9c158 Mon Sep 17 00:00:00 2001 From: Slack Coder Date: Mon, 7 Nov 2022 11:51:01 -0500 Subject: [PATCH] node: prune project routing table Avoid wasting persistent memory by cleaning up stale routing entries when the routing table is considered constrained. Limit pruning to improve resiliency in the face of long intervals between service downtimes. Signed-off-by: Slack Coder --- radicle-node/src/main.rs | 25 ++++++-- radicle-node/src/service.rs | 15 ++++- radicle-node/src/service/config.rs | 22 +++++++ radicle-node/src/service/routing.rs | 60 +++++++++++++++--- radicle-node/src/tests.rs | 95 +++++++++++++++++++++++++++++ 5 files changed, 198 insertions(+), 19 deletions(-) diff --git a/radicle-node/src/main.rs b/radicle-node/src/main.rs index b875fb2d..09ff3d9a 100644 --- a/radicle-node/src/main.rs +++ b/radicle-node/src/main.rs @@ -2,6 +2,8 @@ use std::{env, net, process, thread}; use anyhow::Context as _; +use nakamoto_net::LocalDuration; + use radicle::profile; use radicle_node::crypto::ssh::keystore::MemorySigner; use radicle_node::logger; @@ -13,8 +15,9 @@ type Reactor = nakamoto_net_poll::Reactor; #[derive(Debug)] struct Options { connect: Vec
, - listen: Vec, external_addresses: Vec
, + limits: service::config::Limits, + listen: Vec, } impl Options { @@ -24,6 +27,7 @@ impl Options { let mut parser = lexopt::Parser::from_env(); let mut connect = Vec::new(); let mut external_addresses = Vec::new(); + let mut limits = service::config::Limits::default(); let mut listen = Vec::new(); while let Some(arg) = parser.next()? { @@ -32,14 +36,21 @@ impl Options { let addr = parser.value()?.parse()?; connect.push(addr); } - Long("listen") => { - let addr = parser.value()?.parse()?; - listen.push(addr); - } Long("external-address") => { let addr = parser.value()?.parse()?; external_addresses.push(addr); } + Long("limit-routing-max-age") => { + let secs: u64 = parser.value()?.parse()?; + limits.routing_max_age = LocalDuration::from_secs(secs); + } + Long("limit-routing-max-size") => { + limits.routing_max_size = parser.value()?.parse()?; + } + Long("listen") => { + let addr = parser.value()?.parse()?; + listen.push(addr); + } Long("help") => { println!("usage: radicle-node [--connect ].."); process::exit(0); @@ -49,8 +60,9 @@ impl Options { } Ok(Self { connect, - listen, external_addresses, + limits, + listen, }) } } @@ -77,6 +89,7 @@ fn main() -> anyhow::Result<()> { service: service::Config { connect: options.connect, external_addresses: options.external_addresses, + limits: options.limits, ..service::Config::default() }, listen: options.listen, diff --git a/radicle-node/src/service.rs b/radicle-node/src/service.rs index 2bec7a9a..a1e0be1a 100644 --- a/radicle-node/src/service.rs +++ b/radicle-node/src/service.rs @@ -405,7 +405,7 @@ where if now - self.last_prune >= PRUNE_INTERVAL { debug!("Running 'prune' task..."); - if let Err(err) = self.prune_routing_entries() { + if let Err(err) = self.prune_routing_entries(&now) { error!("Error pruning routing entries: {}", err); } self.reactor.wakeup(PRUNE_INTERVAL); @@ -934,8 +934,17 @@ where Ok(()) } - fn prune_routing_entries(&mut self) -> Result<(), storage::Error> { - // TODO + fn prune_routing_entries(&mut self, now: &LocalTime) -> Result<(), routing::Error> { + let count = self.routing.len()?; + if count <= self.config.limits.routing_max_size { + return Ok(()); + } + + let delta = count - self.config.limits.routing_max_size; + self.routing.prune( + (*now - self.config.limits.routing_max_age).as_secs(), + Some(delta), + )?; Ok(()) } diff --git a/radicle-node/src/service/config.rs b/radicle-node/src/service/config.rs index c48155bb..573227fe 100644 --- a/radicle-node/src/service/config.rs +++ b/radicle-node/src/service/config.rs @@ -1,3 +1,5 @@ +use super::nakamoto::LocalDuration; + use crate::collections::HashSet; use crate::identity::{Id, PublicKey}; use crate::service::filter::Filter; @@ -40,6 +42,24 @@ pub enum RemoteTracking { Allowed(HashSet), } +/// Configuration parameters defining attributes of minima and maxima. +#[derive(Debug, Clone)] +pub struct Limits { + /// Number of routing table entries before we start pruning. + pub routing_max_size: usize, + /// How long to keep a routing table entry before being pruned. + pub routing_max_age: LocalDuration, +} + +impl Default for Limits { + fn default() -> Self { + Self { + routing_max_size: 1000, + routing_max_age: LocalDuration::from_mins(7 * 24 * 60), + } + } +} + /// Service configuration. #[derive(Debug, Clone)] pub struct Config { @@ -58,6 +78,7 @@ pub struct Config { pub relay: bool, /// List of addresses to listen on for protocol connections. pub listen: Vec
, + pub limits: Limits, } impl Default for Config { @@ -70,6 +91,7 @@ impl Default for Config { remote_tracking: RemoteTracking::default(), relay: true, listen: vec![], + limits: Limits::default(), } } } diff --git a/radicle-node/src/service/routing.rs b/radicle-node/src/service/routing.rs index cb8ca128..11aeb469 100644 --- a/radicle-node/src/service/routing.rs +++ b/radicle-node/src/service/routing.rs @@ -16,9 +16,9 @@ pub enum Error { /// An Internal error. #[error("internal error: {0}")] Internal(#[from] sql::Error), - /// Internal clock time overflow. - #[error("the time overflowed")] - TimeOverflow, + /// Internal unit overflow. + #[error("the unit overflowed")] + UnitOverflow, } /// Persistent file storage for a routing table. @@ -61,14 +61,20 @@ pub trait Store { fn get_resources(&self, node_id: &NodeId) -> Result, Error>; /// Get a specific entry. fn entry(&self, id: &Id, node: &NodeId) -> Result, Error>; + /// Checks if any entries are available. + fn is_empty(&self) -> Result { + Ok(self.len()? == 0) + } /// Add a new node seeding the given id. fn insert(&mut self, id: Id, node: NodeId, time: Timestamp) -> Result; /// Remove a node for the given id. fn remove(&mut self, id: &Id, node: &NodeId) -> Result; /// Iterate over all entries in the routing table. fn entries(&self) -> Result>, Error>; + /// Get the total number of routing entries. + fn len(&self) -> Result; /// Prune entries older than the given timestamp. - fn prune(&mut self, oldest: Timestamp) -> Result; + fn prune(&mut self, oldest: Timestamp, limit: Option) -> Result; } impl Store for Table { @@ -113,11 +119,13 @@ impl Store for Table { } fn insert(&mut self, id: Id, node: NodeId, time: Timestamp) -> Result { - let time: i64 = time.try_into().map_err(|_| Error::TimeOverflow)?; + let time: i64 = time.try_into().map_err(|_| Error::UnitOverflow)?; let mut stmt = self.db.prepare( "INSERT INTO routing (resource, node, time) VALUES (?, ?, ?) - ON CONFLICT DO NOTHING", + ON CONFLICT DO UPDATE + SET time = ?3 + WHERE time < ?3", )?; stmt.bind(1, &id)?; @@ -156,11 +164,30 @@ impl Store for Table { Ok(self.db.change_count() > 0) } - fn prune(&mut self, oldest: Timestamp) -> Result { - let oldest: i64 = oldest.try_into().map_err(|_| Error::TimeOverflow)?; - let mut stmt = self.db.prepare("DELETE FROM routing WHERE time < ?")?; + fn len(&self) -> Result { + let stmt = self.db.prepare("SELECT COUNT(1) FROM routing")?; + let count: i64 = stmt + .into_cursor() + .next() + .expect("COUNT will always return a single row")? + .get(0); + let count: usize = count.try_into().map_err(|_| Error::UnitOverflow)?; + Ok(count) + } + fn prune(&mut self, oldest: Timestamp, limit: Option) -> Result { + let oldest: i64 = oldest.try_into().map_err(|_| Error::UnitOverflow)?; + let limit: i64 = limit + .unwrap_or(i64::MAX as usize) + .try_into() + .map_err(|_| Error::UnitOverflow)?; + + let mut stmt = self.db.prepare( + "DELETE FROM routing WHERE rowid IN + (SELECT rowid FROM routing WHERE time < ? LIMIT ?)", + )?; stmt.bind(1, oldest)?; + stmt.bind(2, limit)?; stmt.next()?; Ok(self.db.change_count()) @@ -278,6 +305,19 @@ mod test { assert!(!db.remove(&id, &node).unwrap()); } + #[test] + fn test_len() { + let mut db = Table::open(":memory:").unwrap(); + let ids = arbitrary::vec::(10); + let node = arbitrary::gen(1); + + for id in ids { + db.insert(id, node, LocalTime::now().as_secs()).unwrap(); + } + + assert_eq!(10, db.len().unwrap(), "correct number of rows in table"); + } + #[test] fn test_prune() { let rng = fastrand::Rng::new(); @@ -303,7 +343,7 @@ mod test { } } - let pruned = db.prune(now.as_secs()).unwrap(); + let pruned = db.prune(now.as_secs(), None).unwrap(); assert_eq!(pruned, ids.len() * nodes.len()); for id in &ids { diff --git a/radicle-node/src/tests.rs b/radicle-node/src/tests.rs index b3bb74bc..1dff0dad 100644 --- a/radicle-node/src/tests.rs +++ b/radicle-node/src/tests.rs @@ -7,6 +7,7 @@ use nakamoto_net as nakamoto; use crate::address; use crate::collections::{HashMap, HashSet}; use crate::crypto::test::signer::MockSigner; +use crate::identity::Id; use crate::prelude::{LocalDuration, Timestamp}; use crate::service::config::*; use crate::service::filter::Filter; @@ -227,6 +228,100 @@ fn test_inventory_sync() { } } +#[test] +fn test_inventory_pruning() { + struct Test { + limits: Limits, + /// Number of projects by peer + peer_projects: Vec, + wait_time: LocalDuration, + expected_routing_table_size: usize, + } + let tests = [ + // All zero + Test { + limits: Limits { + routing_max_size: 0, + routing_max_age: LocalDuration::from_secs(0), + }, + peer_projects: vec![10; 5], + wait_time: LocalDuration::from_mins(7 * 24 * 60) + LocalDuration::from_secs(1), + expected_routing_table_size: 0, + }, + // All entries are too young to expire. + Test { + limits: Limits { + routing_max_size: 0, + routing_max_age: LocalDuration::from_mins(7 * 24 * 60), + }, + peer_projects: vec![10; 5], + wait_time: LocalDuration::from_mins(7 * 24 * 60) + LocalDuration::from_secs(1), + expected_routing_table_size: 0, + }, + // All entries remain because the table is unconstrained. + Test { + limits: Limits { + routing_max_size: 50, + routing_max_age: LocalDuration::from_mins(0), + }, + peer_projects: vec![10; 5], + wait_time: LocalDuration::from_mins(7 * 24 * 60) + LocalDuration::from_secs(1), + expected_routing_table_size: 50, + }, + // Some entries are pruned because the table is constrained. + Test { + limits: Limits { + routing_max_size: 25, + routing_max_age: LocalDuration::from_mins(7 * 24 * 60), + }, + peer_projects: vec![10; 5], + wait_time: LocalDuration::from_mins(7 * 24 * 60) + LocalDuration::from_secs(1), + expected_routing_table_size: 25, + }, + ]; + + for test in tests { + let mut alice = Peer::config( + "alice", + Config { + limits: test.limits, + ..Config::default() + }, + [7, 7, 7, 7], + MockStorage::empty(), + address::Book::memory().unwrap(), + MockSigner::default(), + fastrand::Rng::new(), + ); + + let bob = Peer::new("bob", [8, 8, 8, 8], MockStorage::empty()); + + // Tell Alice about the amazing projects available + alice.connect_to(&bob); + for num_projs in test.peer_projects { + alice.receive( + &bob.addr(), + Message::inventory( + InventoryAnnouncement { + inventory: test::arbitrary::vec::(num_projs), + timestamp: bob.clock().timestamp(), + }, + &MockSigner::default(), + ), + ); + } + + // Wait for things to happen + assert!(test.wait_time > PRUNE_INTERVAL, "pruning must be triggered"); + alice.elapse(test.wait_time); + + assert_eq!( + test.expected_routing_table_size, + alice.routing().len().unwrap() + ); + } +} + #[test] fn test_tracking() { let mut alice = Peer::config(