diff --git a/radicle-cli/src/commands/node.rs b/radicle-cli/src/commands/node.rs index e364e770..759c6bff 100644 --- a/radicle-cli/src/commands/node.rs +++ b/radicle-cli/src/commands/node.rs @@ -3,7 +3,7 @@ use std::time; use anyhow::anyhow; -use radicle::node::{Address, Node, NodeId, PeerAddr, ROUTING_DB_FILE}; +use radicle::node::{Address, Node, NodeId, PeerAddr}; use radicle::prelude::Id; use crate::terminal as term; @@ -249,8 +249,7 @@ pub fn run(options: Options, ctx: impl term::Context) -> anyhow::Result<()> { events::run(node, count, timeout)?; } Operation::Routing { rid, nid, json } => { - let store = - radicle::node::routing::Table::reader(profile.home.node().join(ROUTING_DB_FILE))?; + let store = profile.database()?; routing::run(&store, rid, nid, json)?; } Operation::Logs { lines } => control::logs(lines, Some(time::Duration::MAX), &profile)?, diff --git a/radicle-cli/tests/commands.rs b/radicle-cli/tests/commands.rs index b5a48bd3..aff89a8c 100644 --- a/radicle-cli/tests/commands.rs +++ b/radicle-cli/tests/commands.rs @@ -854,9 +854,10 @@ fn rad_clone_connect() { let mut bob = bob.spawn(); // Let Eve know about Alice and Bob having the repo. - eve.routing.insert([&acme], alice.id, now).unwrap(); - eve.routing.insert([&acme], bob.id, now).unwrap(); - eve.addresses + eve.db.routing_mut().insert([&acme], alice.id, now).unwrap(); + eve.db.routing_mut().insert([&acme], bob.id, now).unwrap(); + eve.db + .addresses_mut() .insert( &alice.id, node::Features::SEED, @@ -869,7 +870,8 @@ fn rad_clone_connect() { )], ) .unwrap(); - eve.addresses + eve.db + .addresses_mut() .insert( &bob.id, node::Features::SEED, diff --git a/radicle-httpd/src/api.rs b/radicle-httpd/src/api.rs index 57461807..4753691b 100644 --- a/radicle-httpd/src/api.rs +++ b/radicle-httpd/src/api.rs @@ -62,8 +62,8 @@ impl Context { let delegates = doc.delegates; let issues = issue::Issues::open(&repo)?.counts()?; let patches = patch::Patches::open(&repo)?.counts()?; - let routing = &self.profile.routing()?; - let trackings = routing.count(&id).unwrap_or_default(); + let db = &self.profile.database()?; + let trackings = db.count(&id).unwrap_or_default(); Ok(project::Info { payload, diff --git a/radicle-httpd/src/api/error.rs b/radicle-httpd/src/api/error.rs index 7d6792de..b46c868a 100644 --- a/radicle-httpd/src/api/error.rs +++ b/radicle-httpd/src/api/error.rs @@ -74,9 +74,9 @@ pub enum Error { #[error(transparent)] TrackingStore(#[from] radicle::node::tracking::store::Error), - /// Routing store error. + /// Node database error. #[error(transparent)] - RoutingStore(#[from] radicle::node::routing::Error), + Database(#[from] radicle::node::db::Error), /// Node error. #[error(transparent)] diff --git a/radicle-httpd/src/api/v1/delegates.rs b/radicle-httpd/src/api/v1/delegates.rs index 33fc38dd..8555e015 100644 --- a/radicle-httpd/src/api/v1/delegates.rs +++ b/radicle-httpd/src/api/v1/delegates.rs @@ -38,7 +38,7 @@ async fn delegates_projects_handler( let page = page.unwrap_or(0); let per_page = per_page.unwrap_or(10); let storage = &ctx.profile.storage; - let routing = &ctx.profile.routing()?; + let db = &ctx.profile.database()?; let mut projects = storage .repositories()? .into_iter() @@ -78,7 +78,7 @@ async fn delegates_projects_handler( }; let delegates = id.doc.delegates; - let trackings = routing.count(&id.rid).unwrap_or_default(); + let trackings = db.count(&id.rid).unwrap_or_default(); Some(Info { payload, diff --git a/radicle-httpd/src/api/v1/projects.rs b/radicle-httpd/src/api/v1/projects.rs index 42ff812e..e75b11d6 100644 --- a/radicle-httpd/src/api/v1/projects.rs +++ b/radicle-httpd/src/api/v1/projects.rs @@ -83,7 +83,7 @@ async fn project_root_handler( let page = page.unwrap_or(0); let per_page = per_page.unwrap_or(10); let storage = &ctx.profile.storage; - let routing = &ctx.profile.routing()?; + let db = &ctx.profile.database()?; let mut projects = storage .repositories()? .into_iter() @@ -119,7 +119,7 @@ async fn project_root_handler( return None; }; let delegates = id.doc.delegates; - let trackings = routing.count(&id.rid).unwrap_or_default(); + let trackings = db.count(&id.rid).unwrap_or_default(); Some(Info { payload, diff --git a/radicle-httpd/src/test.rs b/radicle-httpd/src/test.rs index 584e7599..d1431c73 100644 --- a/radicle-httpd/src/test.rs +++ b/radicle-httpd/src/test.rs @@ -19,9 +19,6 @@ use radicle::crypto::{KeyPair, Seed, Signer}; use radicle::git::{raw as git2, RefString}; use radicle::identity::Visibility; use radicle::node; -use radicle::node::address as AddressStore; -use radicle::node::routing as RoutingStore; -use radicle::node::tracking::store as TrackingStore; use radicle::profile; use radicle::profile::Home; use radicle::storage::ReadStorage; @@ -96,15 +93,10 @@ pub fn contributor(dir: &Path) -> Context { fn seed_with_signer(dir: &Path, profile: radicle::Profile, signer: &G) -> Context { const DEFAULT_BRANCH: &str = "master"; - let tracking_db = dir.join("radicle").join("node").join("tracking.db"); - let routing_db = dir.join("radicle").join("node").join("routing.db"); - let addresses_db = dir.join("radicle").join("node").join("addresses.db"); - crate::logger::init().ok(); - TrackingStore::Config::open(tracking_db).unwrap(); - RoutingStore::Table::open(routing_db).unwrap(); - AddressStore::Book::open(addresses_db).unwrap(); + profile.tracking_mut().unwrap(); + profile.database_mut().unwrap(); // Create the database. let workdir = dir.join("hello-world-private"); fs::create_dir_all(&workdir).unwrap(); diff --git a/radicle-node/src/runtime.rs b/radicle-node/src/runtime.rs index a96861c7..c5b65b1c 100644 --- a/radicle-node/src/runtime.rs +++ b/radicle-node/src/runtime.rs @@ -19,7 +19,6 @@ use radicle::node; use radicle::node::address; use radicle::node::address::Store as _; use radicle::node::Handle as _; -use radicle::node::{ADDRESS_DB_FILE, NODE_ANNOUNCEMENT_FILE, ROUTING_DB_FILE, TRACKING_DB_FILE}; use radicle::profile::Home; use radicle::Storage; @@ -42,15 +41,18 @@ 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), + /// A node database error. + #[error("node database error: {0}")] + Database(#[from] node::db::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 address database error. + #[error("address database error: {0}")] + Address(#[from] address::Error), /// An I/O error. #[error("i/o error: {0}")] Io(#[from] io::Error), @@ -137,29 +139,20 @@ impl Runtime { let rng = fastrand::Rng::new(); let clock = LocalTime::now(); let storage = Storage::open(home.storage(), git::UserInfo { alias, key: id })?; - let address_db = node_dir.join(ADDRESS_DB_FILE); - let routing_db = node_dir.join(ROUTING_DB_FILE); - let tracking_db = node_dir.join(TRACKING_DB_FILE); let scope = config.scope; let policy = config.policy; - log::info!(target: "node", "Opening address book {}..", address_db.display()); - let mut addresses = address::Book::open(address_db.clone())?; + log::info!(target: "node", "Opening node database.."); + let mut db: service::Stores<_> = home.database_mut()?.into(); - 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)?; - - log::info!(target: "node", "Opening tracking policy table {}..", tracking_db.display()); - let tracking = tracking::Store::open(tracking_db.clone())?; + log::info!(target: "node", "Opening tracking policy configuration.."); + let tracking = home.tracking_mut()?; let tracking = tracking::Config::new(policy, scope, tracking); log::info!(target: "node", "Default tracking policy set to '{}'", &policy); log::info!(target: "node", "Initializing service ({:?})..", network); - let announcement = if let Some(ann) = fs::read(node_dir.join(NODE_ANNOUNCEMENT_FILE)) + let announcement = if let Some(ann) = fs::read(node_dir.join(node::NODE_ANNOUNCEMENT_FILE)) .ok() .and_then(|ann| NodeAnnouncement::decode(&mut ann.as_slice()).ok()) .and_then(|ann| { @@ -185,13 +178,13 @@ impl Runtime { .expect("Runtime::init: unable to solve proof-of-work puzzle") }; - if config.connect.is_empty() && addresses.is_empty()? { + if config.connect.is_empty() && db.addresses().is_empty()? { log::info!(target: "node", "Address book is empty. Adding bootstrap nodes.."); for (alias, addr) in config.network.bootstrap() { let (id, addr) = addr.into(); - addresses.insert( + db.addresses_mut().insert( &id, radicle::node::Features::SEED, alias, @@ -200,17 +193,15 @@ impl Runtime { [node::KnownAddress::new(addr, address::Source::Bootstrap)], )?; } - log::info!(target: "node", "{} nodes added to address book", addresses.len()?); + log::info!(target: "node", "{} nodes added to address book", db.addresses().len()?); } let emitter: Emitter = Default::default(); let service = service::Service::new( config, clock, - routing, + db, storage.clone(), - addresses, - gossip, tracking, signer.clone(), rng, @@ -246,7 +237,7 @@ impl Runtime { let fetch = worker::FetchConfig { policy, scope, - tracking_db, + tracking_db: home.node().join(node::TRACKING_DB_FILE), limit: FetchLimit::default(), local: nid, expiry: worker::garbage::Expiry::default(), diff --git a/radicle-node/src/service.rs b/radicle-node/src/service.rs index ef0f10da..3abbd477 100644 --- a/radicle-node/src/service.rs +++ b/radicle-node/src/service.rs @@ -21,9 +21,14 @@ use localtime::{LocalDuration, LocalTime}; use log::*; use nonempty::NonEmpty; +use radicle::node; use radicle::node::address; +use radicle::node::address::Store as _; use radicle::node::address::{AddressBook, KnownAddress}; use radicle::node::config::PeerConfig; +use radicle::node::routing::Store as _; +use radicle::node::seed; +use radicle::node::seed::Store as _; use radicle::node::ConnectOptions; use radicle::storage::RepositoryError; @@ -37,6 +42,7 @@ use crate::node::{ }; use crate::prelude::*; use crate::runtime::Emitter; +use crate::service::gossip::Store as _; use crate::service::message::{Announcement, AnnouncementMessage, Info, Ping}; use crate::service::message::{NodeAnnouncement, RefsAnnouncement}; use crate::service::tracking::{store::Write, Scope}; @@ -128,7 +134,11 @@ pub enum Error { #[error(transparent)] Routing(#[from] routing::Error), #[error(transparent)] - Addresses(#[from] address::Error), + Address(#[from] address::Error), + #[error(transparent)] + Database(#[from] node::db::Error), + #[error(transparent)] + Seeds(#[from] seed::Error), #[error(transparent)] Tracking(#[from] tracking::Error), #[error(transparent)] @@ -137,6 +147,11 @@ pub enum Error { Namespaces(#[from] NamespacesError), } +/// A store for all node data. +pub trait Store: address::Store + gossip::Store + routing::Store + seed::Store {} + +impl Store for node::Database {} + /// Function used to query internal service state. pub type QueryState = dyn Fn(&dyn ServiceState) -> Result<(), CommandError> + Send + Sync; @@ -223,23 +238,74 @@ struct FetchState { subscribers: Vec>, } +/// Holds all node stores. +#[derive(Debug)] +pub struct Stores(D); + +impl Stores +where + D: Store, +{ + /// Get the database as a routing store. + pub fn routing(&self) -> &impl routing::Store { + &self.0 + } + + /// Get the database as a routing store, mutably. + pub fn routing_mut(&mut self) -> &mut impl routing::Store { + &mut self.0 + } + + /// Get the database as an address store. + pub fn addresses(&self) -> &impl address::Store { + &self.0 + } + + /// Get the database as an address store, mutably. + pub fn addresses_mut(&mut self) -> &mut impl address::Store { + &mut self.0 + } + + /// Get the database as a gossip store. + pub fn gossip(&self) -> &impl gossip::Store { + &self.0 + } + + /// Get the database as a gossip store, mutably. + pub fn gossip_mut(&mut self) -> &mut impl gossip::Store { + &mut self.0 + } + + /// Get the database as a seed store. + pub fn seeds(&self) -> &impl seed::Store { + &self.0 + } + + /// Get the database as a seed store, mutably. + pub fn seeds_mut(&mut self) -> &mut impl seed::Store { + &mut self.0 + } +} + +impl From for Stores { + fn from(db: D) -> Self { + Self(db) + } +} + /// The node service. #[derive(Debug)] -pub struct Service { +pub struct Service { /// Service configuration. config: Config, /// Our cryptographic signer and key. signer: G, /// Project storage. storage: S, - /// Network routing table. Keeps track of where projects are located. - routing: R, - /// Node address manager. - addresses: A, + /// Node database. + db: Stores, /// Tracking policy configuration. tracking: tracking::Config, - /// State relating to gossip. - gossip: gossip::Store, /// Peer sessions, currently or recently connected. sessions: Sessions, /// Clock. Tells the time. @@ -272,7 +338,7 @@ pub struct Service { emitter: Emitter, } -impl Service +impl Service where G: crypto::Signer, { @@ -287,20 +353,17 @@ where } } -impl Service +impl Service where - R: routing::Store, - A: address::Store, + D: Store, S: ReadStorage + 'static, G: Signer, { pub fn new( config: Config, clock: LocalTime, - routing: R, + db: Stores, storage: S, - addresses: A, - gossip: gossip::Store, tracking: tracking::Config, signer: G, rng: Rng, @@ -312,14 +375,12 @@ where Self { config, storage, - addresses, tracking, signer, rng, node, clock, - routing, - gossip, + db, outbox: Outbox::default(), limiter: RateLimiter::default(), sessions, @@ -382,19 +443,14 @@ where todo!() } - /// Get the address book instance. - pub fn addresses(&self) -> &A { - &self.addresses + /// Get the database. + pub fn database(&self) -> &Stores { + &self.db } - /// Get the mutable address book instance. - pub fn addresses_mut(&mut self) -> &mut A { - &mut self.addresses - } - - /// Get the routing store. - pub fn routing(&self) -> &R { - &self.routing + /// Get the mutable database. + pub fn database_mut(&mut self) -> &mut Stores { + &mut self.db } /// Get the storage instance. @@ -427,9 +483,9 @@ where &mut self.outbox } - /// Lookup a project, both locally and in the routing table. + /// Lookup a repository, both locally and in the routing table. pub fn lookup(&self, rid: Id) -> Result { - let remote = self.routing.get(&rid)?.iter().cloned().collect(); + let remote = self.db.routing().get(&rid)?.iter().cloned().collect(); Ok(Lookup { local: self.storage.get(rid)?, @@ -440,12 +496,14 @@ where pub fn initialize(&mut self, time: LocalTime) -> Result<(), Error> { debug!(target: "service", "Init @{}", time.as_millis()); + let nid = self.node_id(); self.start_time = time; // Ensure that our local node is in our address database. - self.addresses + self.db + .addresses_mut() .insert( - &self.node_id(), + &nid, self.node.features, self.node.alias.clone(), self.node.work(), @@ -466,12 +524,11 @@ where // all of it. It can happen that inventory is not properly tracked if for eg. the // user creates a new repository while the node is stopped. let rids = self.storage.inventory()?; - self.routing - .insert(&rids, self.node_id(), time.as_millis())?; + self.db.routing_mut().insert(&rids, nid, time.as_millis())?; - let nid = self.node_id(); let announced = self - .addresses + .db + .seeds() .seeded_by(&nid)? .collect::, _>>()?; for rid in rids { @@ -493,15 +550,19 @@ where } // Make sure our local node's sync status is up to date with storage. debug!(target: "service", "Saving local sync status for {rid}.."); - self.addresses - .synced(&rid, &nid, updated_at.oid, updated_at.timestamp.as_millis())?; + self.db.seeds_mut().synced( + &rid, + &nid, + updated_at.oid, + updated_at.timestamp.as_millis(), + )?; // If we got here, it likely means a repo was updated while the node was stopped. // Therefore, we pre-load a refs announcement for this repo, so that it is included in // the historical gossip messages when a node connects and subscribes to this repo. if let Ok((ann, _)) = self.refs_announcement_for(rid, [nid]) { debug!(target: "service", "Adding refs announcement for {rid} to historical gossip messages.."); - self.gossip.announced(&nid, &ann)?; + self.db.gossip_mut().announced(&nid, &ann)?; } } @@ -566,7 +627,8 @@ where error!(target: "service", "Error pruning routing entries: {err}"); } if let Err(err) = self - .gossip + .db + .gossip_mut() .prune((now - self.config.limits.gossip_max_age).as_millis()) { error!(target: "service", "Error pruning gossip entries: {err}"); @@ -925,7 +987,7 @@ where peer.to_connected(self.clock); self.outbox.write_all(peer, msgs); - if let Err(e) = self.addresses.connected(&remote, &peer.addr, now) { + if let Err(e) = self.db.addresses_mut().connected(&remote, &peer.addr, now) { error!(target: "service", "Error updating address book with connection: {e}"); } } @@ -1000,7 +1062,8 @@ where debug!(target: "service", "Dropping peer {remote}.."); if let Err(e) = - self.addresses + self.db + .addresses_mut() .disconnected(&remote, &session.addr, reason.is_transient()) { error!(target: "service", "Error updating address store: {e}"); @@ -1068,7 +1131,7 @@ where // announcement timestamp, but before the other announcements. In that case, we simply // ignore all announcements of that node until we get a node announcement. if let AnnouncementMessage::Inventory(_) | AnnouncementMessage::Refs(_) = message { - match self.addresses.get(announcer) { + match self.db.addresses().get(announcer) { Ok(node) => { if node.is_none() { debug!(target: "service", "Ignoring announcement from unknown node {announcer}"); @@ -1083,7 +1146,7 @@ where } // Discard announcement messages we've already seen, otherwise update our last seen time. - match self.gossip.announced(announcer, announcement) { + match self.db.gossip_mut().announced(announcer, announcement) { Ok(fresh) => { if !fresh { trace!(target: "service", "Ignoring stale inventory announcement from {announcer} (t={})", self.time()); @@ -1162,7 +1225,8 @@ where // We update inventories when receiving ref announcements, as these could come // from a new repository being initialized. if let Ok(result) = - self.routing + self.db + .routing_mut() .insert([&message.rid], *announcer, message.timestamp) { if let &[(_, InsertResult::SeedAdded)] = result.as_slice() { @@ -1176,10 +1240,12 @@ where // Update sync status for this repo. if let Some(refs) = message.refs.iter().find(|r| &r.remote == self.nid()) { - match self - .addresses - .synced(&message.rid, announcer, refs.at, message.timestamp) - { + match self.db.seeds_mut().synced( + &message.rid, + announcer, + refs.at, + message.timestamp, + ) { Ok(updated) => { if updated { debug!( @@ -1295,7 +1361,7 @@ where return Ok(relay); } - match self.addresses.insert( + match self.db.addresses_mut().insert( announcer, *features, ann.alias.clone(), @@ -1426,7 +1492,8 @@ where (session::State::Connected { .. }, Message::Subscribe(subscribe)) => { // Filter announcements by interest. match self - .gossip + .db + .gossip() .filtered(&subscribe.filter, subscribe.since, subscribe.until) { Ok(anns) => { @@ -1507,7 +1574,7 @@ where // // If this is our first connection to the network, we just ask for a fixed backlog // of messages to get us started. - let since = match self.gossip.last() { + let since = match self.db.gossip().last() { Ok(Some(last)) => last - MAX_TIME_DELTA.as_millis() as Timestamp, Ok(None) => (*now - INITIAL_SUBSCRIBE_BACKLOG_DELTA).as_millis() as Timestamp, Err(e) => { @@ -1545,7 +1612,7 @@ where let mut synced = SyncedRouting::default(); let included: HashSet<&Id> = HashSet::from_iter(inventory); - for (rid, result) in self.routing.insert(inventory, from, timestamp)? { + for (rid, result) in self.db.routing_mut().insert(inventory, from, timestamp)? { match result { InsertResult::SeedAdded => { info!(target: "service", "Routing table updated for {rid} with seed {from}"); @@ -1565,9 +1632,9 @@ where InsertResult::NotUpdated => {} } } - for rid in self.routing.get_resources(&from)?.into_iter() { + for rid in self.db.routing().get_resources(&from)?.into_iter() { if !included.contains(&rid) { - if self.routing.remove(&rid, &from)? { + if self.db.routing_mut().remove(&rid, &from)? { synced.removed.push(rid); self.emitter.emit(Event::SeedDropped { rid, nid: from }); } @@ -1622,7 +1689,8 @@ where // the node was stopped. if let Some(refs) = refs.iter().find(|r| r.remote == ann.node) { if let Err(e) = self - .addresses + .db + .seeds_mut() .synced(&rid, &ann.node, refs.at, ann.timestamp()) { error!(target: "service", "Error updating sync status for local node: {e}"); @@ -1634,7 +1702,7 @@ where // Only announce to peers who are allowed to view this repo. doc.is_visible_to(&p.id) }), - &mut self.gossip, + self.db.gossip_mut(), ); Ok(refs) @@ -1682,8 +1750,9 @@ where return false; } let persistent = self.config.is_persistent(&nid); + let time = self.time(); - if let Err(e) = self.addresses.attempted(&nid, &addr, self.time()) { + if let Err(e) = self.db.addresses_mut().attempted(&nid, &addr, time) { error!(target: "service", "Error updating address book with connection attempt: {e}"); } self.sessions.insert( @@ -1709,7 +1778,7 @@ where // our own refs. if let Ok(repo) = self.storage.repository(*rid) { if let Ok(local) = RefsAt::new(&repo, self.node_id()) { - for seed in self.addresses.seeds(rid)? { + for seed in self.db.seeds().seeds_for(rid)? { let seed = seed?; let state = self.sessions.get(&seed.nid).map(|s| s.state.clone()); let synced = if local.at == seed.synced_at.oid { @@ -1730,7 +1799,7 @@ where // Then, add peers we know about but have no information about the sync status. // These peers have announced that they track the repository via an inventory // announcement, but we haven't received any ref announcements from them. - for nid in self.routing.get(rid)? { + for nid in self.db.routing().get(rid)? { if nid == self.node_id() { continue; } @@ -1738,7 +1807,7 @@ where // We already have a richer entry for this node. continue; } - let addrs = self.addresses.addresses(&nid)?; + let addrs = self.db.addresses().addresses_of(&nid)?; let state = self.sessions.get(&nid).map(|s| s.state.clone()); seeds.insert(Seed::new(nid, addrs, state, None)); @@ -1773,19 +1842,19 @@ where self.outbox.announce( msg.signed(&self.signer), self.sessions.connected().map(|(_, p)| p), - &mut self.gossip, + self.db.gossip_mut(), ); Ok(()) } fn prune_routing_entries(&mut self, now: &LocalTime) -> Result<(), routing::Error> { - let count = self.routing.len()?; + let count = self.db.routing().len()?; if count <= self.config.limits.routing_max_size { return Ok(()); } let delta = count - self.config.limits.routing_max_size; - self.routing.prune( + self.db.routing_mut().prune( (*now - self.config.limits.routing_max_age).as_millis(), Some(delta), )?; @@ -1825,7 +1894,7 @@ where /// Get a list of peers available to connect to. fn available_peers(&mut self) -> HashMap> { - match self.addresses.entries() { + match self.db.addresses().entries() { Ok(entries) => { // Nb. we don't want to connect to any peers that already have a session with us, // even if it's in a disconnected state. Those sessions are re-attempted automatically. @@ -1969,9 +2038,9 @@ pub trait ServiceState { fn config(&self) -> &Config; } -impl ServiceState for Service +impl ServiceState for Service where - R: routing::Store, + D: routing::Store, G: Signer, S: ReadStorage, { diff --git a/radicle-node/src/service/gossip.rs b/radicle-node/src/service/gossip.rs index 02cd159f..51253866 100644 --- a/radicle-node/src/service/gossip.rs +++ b/radicle-node/src/service/gossip.rs @@ -3,7 +3,7 @@ pub mod store; use super::*; pub use store::Error; -pub use store::GossipStore as Store; +pub use store::Store; pub fn node(config: &Config, timestamp: Timestamp) -> NodeAnnouncement { let features = config.features(); diff --git a/radicle-node/src/service/gossip/store.rs b/radicle-node/src/service/gossip/store.rs index a7c35356..65eb72e2 100644 --- a/radicle-node/src/service/gossip/store.rs +++ b/radicle-node/src/service/gossip/store.rs @@ -1,10 +1,10 @@ -use std::{fmt, io, path::Path}; +use std::{fmt, io}; use radicle::crypto::Signature; use sqlite as sql; use thiserror::Error; -use crate::node::NodeId; +use crate::node::{Database, NodeId}; use crate::prelude::{Filter, Timestamp}; use crate::service::message::{ Announcement, AnnouncementMessage, InventoryAnnouncement, NodeAnnouncement, RefsAnnouncement, @@ -14,39 +14,42 @@ 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), } +/// A database that has access to historical gossip messages. /// 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().with_read_write().with_full_mutex(), - )?; - - Ok(Self { db }) - } - +pub trait Store { /// Prune announcements older than the cutoff time. - pub fn prune(&mut self, cutoff: Timestamp) -> Result { + fn prune(&mut self, cutoff: Timestamp) -> Result; + + /// Get the timestamp of the last announcement in the store. + fn last(&self) -> Result, Error>; + + /// Process an announcement for the given node. + /// Returns `true` if the timestamp was updated or the announcement wasn't there before. + fn announced(&mut self, nid: &NodeId, ann: &Announcement) -> Result; + + /// Get all the latest gossip messages of all nodes, filtered by inventory filter and + /// announcement timestamps. + /// + /// # Panics + /// + /// Panics if `from` > `to`. + /// + fn filtered<'a>( + &'a self, + filter: &'a Filter, + from: Timestamp, + to: Timestamp, + ) -> Result> + 'a>, Error>; +} + +impl Store for Database { + fn prune(&mut self, cutoff: Timestamp) -> Result { let mut stmt = self .db .prepare("DELETE FROM `announcements` WHERE timestamp < ?1")?; @@ -57,8 +60,7 @@ impl GossipStore { Ok(self.db.change_count()) } - /// Get the timestamp of the last announcement in the store. - pub fn last(&self) -> Result, Error> { + fn last(&self) -> Result, Error> { let stmt = self .db .prepare("SELECT MAX(timestamp) AS latest FROM `announcements`")?; @@ -71,9 +73,7 @@ impl GossipStore { Ok(None) } - /// 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 { + 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) @@ -107,19 +107,12 @@ impl GossipStore { 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>( + fn filtered<'a>( &'a self, filter: &'a Filter, from: Timestamp, to: Timestamp, - ) -> Result> + 'a, Error> { + ) -> Result> + 'a>, Error> { let mut stmt = self.db.prepare( "SELECT node, type, message, signature, timestamp FROM announcements @@ -131,41 +124,42 @@ impl GossipStore { 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"); + Ok(Box::new( + 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); + debug_assert_eq!(timestamp, message.timestamp() as i64); - Ok(Announcement { - node, - message, - signature, + Ok(Announcement { + node, + message, + signature, + }) }) - }) - .filter(|ann| match ann { - Ok(a) => a.matches(filter), - Err(_) => true, - })) + .filter(|ann| match ann { + Ok(a) => a.matches(filter), + Err(_) => true, + }), + )) } } diff --git a/radicle-node/src/service/io.rs b/radicle-node/src/service/io.rs index f4ad8ee1..b48a2279 100644 --- a/radicle-node/src/service/io.rs +++ b/radicle-node/src/service/io.rs @@ -68,7 +68,7 @@ impl Outbox { &mut self, ann: Announcement, peers: impl Iterator, - gossip: &mut gossip::Store, + gossip: &mut impl gossip::Store, ) { // Store our announcement so that it can be retrieved from us later, just like // announcements we receive from peers. diff --git a/radicle-node/src/test/environment.rs b/radicle-node/src/test/environment.rs index 72afdc04..206b281f 100644 --- a/radicle-node/src/test/environment.rs +++ b/radicle-node/src/test/environment.rs @@ -18,11 +18,10 @@ use radicle::crypto::{KeyPair, Seed, Signer}; use radicle::git; use radicle::git::refname; use radicle::identity::{Id, Visibility}; -use radicle::node::address::Book; -use radicle::node::routing; use radicle::node::routing::Store; use radicle::node::tracking::store as tracking; -use radicle::node::{Alias, ADDRESS_DB_FILE, ROUTING_DB_FILE, TRACKING_DB_FILE}; +use radicle::node::Database; +use radicle::node::{Alias, TRACKING_DB_FILE}; use radicle::node::{ConnectOptions, Handle as _}; use radicle::profile; use radicle::profile::Home; @@ -84,20 +83,15 @@ impl Environment { let tracking_db = profile.home.node().join(TRACKING_DB_FILE); let tracking = tracking::Config::open(tracking_db).unwrap(); - - let routing_db = profile.home.node().join(ROUTING_DB_FILE); - let routing = routing::Table::open(routing_db).unwrap(); - - let addresses_db = profile.home.node().join(ADDRESS_DB_FILE); - let addresses = Book::open(addresses_db).unwrap(); + let db = profile.database_mut().unwrap(); + let db = service::Stores::from(db); Node { id: *profile.id(), home: profile.home, config, signer, - addresses, - routing, + db, tracking, storage: profile.storage, } @@ -129,8 +123,7 @@ impl Environment { .unwrap(); tracking::Config::open(tracking_db).unwrap(); - let addresses_db = home.node().join(ADDRESS_DB_FILE); - Book::open(addresses_db).unwrap(); + home.database_mut().unwrap(); // Just create the database. transport::local::register(storage.clone()); keystore.store(keypair.clone(), "radicle", None).unwrap(); @@ -155,8 +148,7 @@ pub struct Node { pub signer: G, pub storage: Storage, pub config: Config, - pub addresses: Book, - pub routing: routing::Table, + pub db: service::Stores, pub tracking: tracking::Config, } @@ -217,7 +209,7 @@ impl NodeHandle { /// Get routing table entries. pub fn routing(&self) -> impl Iterator { - radicle::node::routing::Table::reader(self.home.node().join(radicle::node::ROUTING_DB_FILE)) + Database::reader(self.home.node().join(node::NODE_DB_FILE)) .unwrap() .entries() .unwrap() @@ -356,9 +348,9 @@ impl Node { }, ) .unwrap(); - let addresses = home.addresses_mut().unwrap(); let tracking = home.tracking_mut().unwrap(); - let routing = home.routing_mut().unwrap(); + let db = home.database_mut().unwrap(); + let db = service::Stores::from(db); log::debug!(target: "test", "Node::init {}: {}", config.alias, signer.public_key()); Self { @@ -367,9 +359,8 @@ impl Node { signer, storage, config, - addresses, + db, tracking, - routing, } } } diff --git a/radicle-node/src/test/peer.rs b/radicle-node/src/test/peer.rs index bcc6e20a..60dc9cb0 100644 --- a/radicle-node/src/test/peer.rs +++ b/radicle-node/src/test/peer.rs @@ -7,7 +7,8 @@ use std::str::FromStr; use log::*; use radicle::identity::Visibility; -use radicle::node::address::Store; +use radicle::node::address::Store as _; +use radicle::node::Database; use radicle::node::{address, Alias, ConnectOptions}; use radicle::rad; use radicle::storage::refs::RefsAt; @@ -18,7 +19,6 @@ use crate::crypto::test::signer::MockSigner; use crate::crypto::Signer; use crate::identity::Id; use crate::node; -use crate::node::routing; use crate::prelude::*; use crate::runtime::Emitter; use crate::service; @@ -36,7 +36,7 @@ use crate::Link; use crate::{LocalDuration, LocalTime}; /// Service instantiation used for testing. -pub type Service = service::Service; +pub type Service = service::Service; #[derive(Debug)] pub struct Peer { @@ -100,8 +100,7 @@ where pub struct Config { pub config: service::Config, - pub addrs: address::Book, - pub gossip: gossip::Store, + pub db: Stores, pub local_time: LocalTime, pub policy: Policy, pub scope: Scope, @@ -115,13 +114,13 @@ impl Default for Config { 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(); + let db = Database::open(tmp.path().join(node::NODE_DB_FILE)) + .unwrap() + .into(); Config { config: service::Config::test(Alias::from_str("mocky").unwrap()), - addrs, - gossip, + db, local_time: LocalTime::now(), policy: Policy::default(), scope: Scope::default(), @@ -163,7 +162,6 @@ where storage: S, mut config: Config, ) -> Self { - let routing = routing::Table::memory().unwrap(); let tracking = tracking::Store::::memory().unwrap(); let mut tracking = tracking::Config::new(config.policy, config.scope, tracking); let id = *config.signer.public_key(); @@ -181,10 +179,8 @@ where let service = Service::new( config.config, config.local_time, - routing, + config.db, storage, - config.addrs, - config.gossip, tracking, config.signer, config.rng.clone(), @@ -236,8 +232,9 @@ where pub fn import_addresses<'a>(&mut self, peers: impl IntoIterator) { let timestamp = self.timestamp(); for peer in peers.into_iter() { - let known_address = address::KnownAddress::new(peer.address(), address::Source::Peer); + let known_address = node::KnownAddress::new(peer.address(), address::Source::Peer); self.service + .database_mut() .addresses_mut() .insert( &peer.node_id(), diff --git a/radicle-node/src/tests.rs b/radicle-node/src/tests.rs index b640b96c..58829c04 100644 --- a/radicle-node/src/tests.rs +++ b/radicle-node/src/tests.rs @@ -273,7 +273,7 @@ fn test_inventory_sync() { ); for proj in &projs { - let seeds = alice.routing().get(proj).unwrap(); + let seeds = alice.database().routing().get(proj).unwrap(); assert!(seeds.contains(&bob.node_id())); } } @@ -382,7 +382,7 @@ fn test_inventory_pruning() { assert_eq!( test.expected_routing_table_size, - alice.routing().len().unwrap() + alice.database().routing().len().unwrap() ); } } diff --git a/radicle-node/src/wire/protocol.rs b/radicle-node/src/wire/protocol.rs index 19757b2e..7ef30a61 100644 --- a/radicle-node/src/wire/protocol.rs +++ b/radicle-node/src/wire/protocol.rs @@ -22,7 +22,7 @@ use netservices::{NetConnection, NetProtocol, NetReader, NetWriter}; use reactor::Timestamp; use radicle::collections::RandomMap; -use radicle::node::{address, routing, NodeId}; +use radicle::node::NodeId; use radicle::storage::WriteStorage; use crate::crypto::Signer; @@ -316,9 +316,9 @@ impl Peers { } /// Wire protocol implementation for a set of peers. -pub struct Wire { +pub struct Wire { /// Backing service instance. - service: Service, + service: Service, /// Worker pool interface. worker: chan::Sender, /// Used for authentication. @@ -331,15 +331,14 @@ pub struct Wire { proxy: net::SocketAddr, } -impl Wire +impl Wire where - R: routing::Store, - S: address::Store, - W: WriteStorage + 'static, + D: service::Store, + S: WriteStorage + 'static, G: Signer + Ecdh, { pub fn new( - mut service: Service, + mut service: Service, worker: chan::Sender, signer: G, proxy: net::SocketAddr, @@ -450,11 +449,10 @@ where } } -impl reactor::Handler for Wire +impl reactor::Handler for Wire where - R: routing::Store + Send, - S: address::Store + Send, - W: WriteStorage + Send + 'static, + D: service::Store + Send, + S: WriteStorage + Send + 'static, G: Signer + Ecdh + Clone + Send, { type Listener = NetAccept>; @@ -742,11 +740,10 @@ where } } -impl Iterator for Wire +impl Iterator for Wire where - R: routing::Store, - S: address::Store, - W: WriteStorage + 'static, + D: service::Store, + S: WriteStorage + 'static, G: Signer + Ecdh, { type Item = Action; diff --git a/radicle/src/lib.rs b/radicle/src/lib.rs index 82f73b0e..7fe023be 100644 --- a/radicle/src/lib.rs +++ b/radicle/src/lib.rs @@ -9,6 +9,7 @@ extern crate amplify; extern crate radicle_git_ext as git_ext; mod canonical; + pub mod cli; pub mod cob; pub mod collections; diff --git a/radicle/src/node.rs b/radicle/src/node.rs index b4f7378e..af777823 100644 --- a/radicle/src/node.rs +++ b/radicle/src/node.rs @@ -2,8 +2,10 @@ mod features; pub mod address; pub mod config; +pub mod db; pub mod events; pub mod routing; +pub mod seed; pub mod tracking; use std::collections::{BTreeSet, HashMap, HashSet}; @@ -27,11 +29,13 @@ use crate::profile; use crate::storage::refs::RefsAt; use crate::storage::RefUpdate; -pub use address::{KnownAddress, SyncedAt}; +pub use address::KnownAddress; pub use config::Config; pub use cyphernet::addr::{HostName, PeerAddr}; +pub use db::Database; pub use events::{Event, Events}; pub use features::Features; +pub use seed::SyncedAt; /// Default name for control socket file. pub const DEFAULT_SOCKET_NAME: &str = "control.sock"; @@ -41,10 +45,8 @@ pub const DEFAULT_PORT: u16 = 8776; pub const DEFAULT_TIMEOUT: time::Duration = time::Duration::from_secs(9); /// Maximum length in bytes of a node alias. pub const MAX_ALIAS_LENGTH: usize = 32; -/// Filename of routing table database under the node directory. -pub const ROUTING_DB_FILE: &str = "routing.db"; -/// Filename of address database under the node directory. -pub const ADDRESS_DB_FILE: &str = "addresses.db"; +/// Filename of node database under the node directory. +pub const NODE_DB_FILE: &str = "node.db"; /// Filename of tracking table database under the node directory. pub const TRACKING_DB_FILE: &str = "tracking.db"; /// Filename of last node announcement, when running in debug mode. @@ -462,13 +464,18 @@ impl Session { } } +/// A seed for some repository, with metadata about its status. #[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)] #[serde(rename_all = "camelCase")] pub struct Seed { + /// The Node ID. pub nid: NodeId, + /// Known addresses for this seed. pub addrs: Vec, + /// The seed's session state, if any. #[serde(default, skip_serializing_if = "Option::is_none")] pub state: Option, + /// The seed's sync status, if any. #[serde(default, skip_serializing_if = "Option::is_none")] pub sync: Option, } @@ -1069,18 +1076,6 @@ pub trait AliasStore { fn alias(&self, nid: &NodeId) -> Option; } -impl AliasStore for &T { - fn alias(&self, nid: &NodeId) -> Option { - (*self).alias(nid) - } -} - -impl AliasStore for Box { - fn alias(&self, nid: &NodeId) -> Option { - self.deref().alias(nid) - } -} - impl AliasStore for HashMap { fn alias(&self, nid: &NodeId) -> Option { self.get(nid).map(ToOwned::to_owned) diff --git a/radicle/src/node/address.rs b/radicle/src/node/address.rs index f3c6ceb3..3d9290ed 100644 --- a/radicle/src/node/address.rs +++ b/radicle/src/node/address.rs @@ -1,13 +1,190 @@ -mod store; -mod types; +pub mod store; +pub use store::{Error, Store}; -pub use store::*; -pub use types::*; +use std::cell::RefCell; +use std::ops::{Deref, DerefMut}; +use std::{hash, net}; use cyphernet::addr::HostName; -use std::net; +use localtime::LocalTime; +use nonempty::NonEmpty; -use crate::node::Address; +use crate::collections::RandomMap; +use crate::node::{Address, Alias}; +use crate::prelude::Timestamp; +use crate::{node, profile}; + +/// A map with the ability to randomly select values. +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] +#[serde(transparent)] +pub struct AddressBook { + inner: RandomMap, + #[serde(skip)] + rng: RefCell, +} + +impl AddressBook { + /// Create a new address book. + pub fn new(rng: fastrand::Rng) -> Self { + Self { + inner: RandomMap::with_hasher(rng.clone().into()), + rng: RefCell::new(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.borrow_mut().usize(..pairs.len()); + let pair = pairs[ix]; // Can't fail. + + Some(pair) + } else { + None + } + } + + /// Return a new address book with the given RNG. + pub fn with(self, rng: fastrand::Rng) -> Self { + Self { + inner: self.inner, + rng: RefCell::new(rng), + } + } +} + +impl AddressBook { + /// Return a shuffled iterator. + pub fn shuffled(&self) -> std::vec::IntoIter<(&K, &V)> { + let mut items = self.inner.iter().collect::>(); + items.sort_by_key(|(k, _)| *k); + self.rng.borrow_mut().shuffle(&mut items); + + items.into_iter() + } + + /// Turn this object into a shuffled iterator. + pub fn into_shuffled(self) -> impl Iterator { + let mut items = self.inner.into_iter().collect::>(); + items.sort_by_key(|(k, _)| *k); + self.rng.borrow_mut().shuffle(&mut items); + + items.into_iter() + } + + /// Cycle through the keys at random. The random cycle repeats ad-infintum. + pub fn cycle(&self) -> impl Iterator { + self.shuffled().map(|(k, _)| k).cycle() + } +} + +impl FromIterator<(K, V)> for AddressBook { + fn from_iter>(iter: T) -> Self { + let rng = profile::env::rng(); + let mut inner = RandomMap::with_hasher(rng.clone().into()); + + for (k, v) in iter { + inner.insert(k, v); + } + Self { + inner, + rng: RefCell::new(rng), + } + } +} + +impl Deref for AddressBook { + type Target = RandomMap; + + fn deref(&self) -> &Self::Target { + &self.inner + } +} + +impl DerefMut for AddressBook { + fn deref_mut(&mut self) -> &mut Self::Target { + &mut self.inner + } +} + +/// Node public data. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Node { + /// Advertized alias. + pub alias: Alias, + /// Advertized features. + pub features: node::Features, + /// Advertized addresses + pub addrs: Vec, + /// Proof-of-work included in node announcement. + pub pow: u32, + /// When this data was published. + pub timestamp: Timestamp, +} + +/// A known address. +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde(rename_all = "camelCase")] +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. + #[serde(with = "crate::serde_ext::localtime::option::time")] + pub last_success: Option, + /// Last time this address was tried. + #[serde(with = "crate::serde_ext::localtime::option::time")] + pub last_attempt: Option, + /// Whether this address has been banned. + pub banned: bool, +} + +impl KnownAddress { + /// Create a new known address. + pub fn new(addr: Address, source: Source) -> Self { + Self { + addr, + source, + last_success: None, + last_attempt: None, + banned: false, + } + } +} + +/// Address source. Specifies where an address originated from. +#[derive(Debug, Copy, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde(rename_all = "camelCase")] +pub enum Source { + /// An address that was shared by another peer. + Peer, + /// An bootstrap node address. + Bootstrap, + /// 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::Bootstrap => write!(f, "Bootstrap"), + Self::Imported => write!(f, "Imported"), + } + } +} /// Address type. #[repr(u8)] diff --git a/radicle/src/node/address/store.rs b/radicle/src/node/address/store.rs index 1d750ac2..a89d326d 100644 --- a/radicle/src/node/address/store.rs +++ b/radicle/src/node/address/store.rs @@ -1,93 +1,64 @@ -#![allow(clippy::type_complexity)] -use std::path::Path; use std::str::FromStr; -use std::{fmt, io}; use localtime::LocalTime; use sqlite as sql; use thiserror::Error; -use crate::git::Oid; use crate::node; -use crate::node::address::{KnownAddress, Source}; -use crate::node::{Address, Alias, AliasError, AliasStore, NodeId}; -use crate::prelude::{Id, Timestamp}; +use crate::node::address::{AddressType, KnownAddress, Node, Source}; +use crate::node::{Address, Alias, AliasError, AliasStore, Database, NodeId}; +use crate::prelude::Timestamp; use crate::sql::transaction; -use super::types; -use super::{AddressType, SyncedAt}; - #[derive(Error, Debug)] pub enum Error { - /// I/O error. - #[error("i/o error: {0}")] - Io(#[from] io::Error), - #[error("alias error: {0}")] - InvalidAlias(#[from] AliasError), /// An Internal error. #[error("internal error: {0}")] Internal(#[from] sql::Error), + #[error("alias error: {0}")] + InvalidAlias(#[from] AliasError), } -/// A file-backed address book. -pub struct Book { - db: sql::Connection, +/// Address store. +/// +/// Used to store node addresses and metadata. +pub trait Store { + /// Get the information we have about a node. + fn get(&self, id: &NodeId) -> Result, Error>; + /// Get the addresses of a node. + fn addresses_of(&self, node: &NodeId) -> Result, Error>; + /// Insert a node with associated addresses into the store. + /// + /// Returns `true` if the node or addresses were updated, and `false` otherwise. + fn insert( + &mut self, + node: &NodeId, + features: node::Features, + alias: Alias, + pow: u32, + timestamp: Timestamp, + addrs: impl IntoIterator, + ) -> Result; + /// Remove a node from the store. + fn remove(&mut self, id: &NodeId) -> Result; + /// Returns the number of addresses. + fn len(&self) -> Result; + /// Returns true if there are no addresses. + fn is_empty(&self) -> Result { + self.len().map(|l| l == 0) + } + /// Get the address entries in the store. + fn entries(&self) -> Result>, Error>; + /// Mark a node as attempted at a certain time. + fn attempted(&self, nid: &NodeId, addr: &Address, time: Timestamp) -> Result<(), Error>; + /// Mark a node as successfully connected at a certain time. + fn connected(&self, nid: &NodeId, addr: &Address, time: Timestamp) -> Result<(), Error>; + /// Mark a node as disconnected. + fn disconnected(&mut self, nid: &NodeId, addr: &Address, transient: bool) -> Result<(), Error>; } -impl fmt::Debug for Book { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - write!(f, "Book(..)") - } -} - -impl From for Book { - fn from(db: sql::Connection) -> Self { - Self { db } - } -} - -impl Book { - const SCHEMA: &'static str = include_str!("schema.sql"); - const PRAGMA: &'static str = "PRAGMA foreign_keys = ON"; - - /// 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_with_flags( - path, - sqlite::OpenFlags::new() - .with_create() - .with_read_write() - .with_full_mutex(), - )?; - db.execute(Self::PRAGMA)?; - db.execute(Self::SCHEMA)?; - - Ok(Self { db }) - } - - /// Same as [`Self::open`], but in read-only mode. This is useful to have multiple - /// open databases, as no locking is required. - pub fn reader>(path: P) -> Result { - let db = sql::Connection::open_with_flags(path, sqlite::OpenFlags::new().with_read_only())?; - db.execute(Self::PRAGMA)?; - db.execute(Self::SCHEMA)?; - - Ok(Self { db }) - } - - /// Create a new in-memory address book. - pub fn memory() -> Result { - let db = sql::Connection::open(":memory:")?; - db.execute(Self::PRAGMA)?; - db.execute(Self::SCHEMA)?; - - Ok(Self { db }) - } -} - -impl Store for Book { - fn get(&self, node: &NodeId) -> Result, Error> { +impl Store for Database { + fn get(&self, node: &NodeId) -> Result, Error> { let mut stmt = self .db .prepare("SELECT features, alias, pow, timestamp FROM nodes WHERE id = ?")?; @@ -99,9 +70,9 @@ impl Store for Book { let alias = Alias::from_str(row.read::<&str, _>("alias"))?; let timestamp = row.read::("timestamp") as Timestamp; let pow = row.read::("pow") as u32; - let addrs = self.addresses(node)?; + let addrs = self.addresses_of(node)?; - Ok(Some(types::Node { + Ok(Some(Node { features, alias, pow, @@ -113,7 +84,7 @@ impl Store for Book { } } - fn addresses(&self, node: &NodeId) -> Result, Error> { + fn addresses_of(&self, node: &NodeId) -> Result, Error> { let mut addrs = Vec::new(); let mut stmt = self.db.prepare( "SELECT type, value, source, last_attempt, last_success, banned FROM addresses WHERE node = ?", @@ -211,90 +182,6 @@ impl Store for Book { Ok(self.db.change_count() > 0) } - fn synced( - &mut self, - rid: &Id, - nid: &NodeId, - at: Oid, - timestamp: Timestamp, - ) -> Result { - let mut stmt = self.db.prepare( - "INSERT INTO `repo-sync-status` (repo, node, head, timestamp) - VALUES (?1, ?2, ?3, ?4) - ON CONFLICT DO UPDATE - SET head = ?3, timestamp = ?4 - WHERE timestamp < ?4", - )?; - stmt.bind((1, rid))?; - stmt.bind((2, nid))?; - stmt.bind((3, at.to_string().as_str()))?; - stmt.bind((4, timestamp as i64))?; - stmt.next()?; - - Ok(self.db.change_count() > 0) - } - - fn seeds( - &self, - rid: &Id, - ) -> Result> + '_>, Error> { - let mut stmt = self.db.prepare( - "SELECT node, head, timestamp - FROM `repo-sync-status` - WHERE repo = ?", - )?; - stmt.bind((1, rid))?; - - Ok(Box::new(stmt.into_iter().map(|row| { - let row = row?; - let nid = row.try_read::("node")?; - let oid = row.try_read::<&str, _>("head")?; - let oid = Oid::from_str(oid).map_err(|e| { - Error::Internal(sql::Error { - code: None, - message: Some(format!("sql: invalid oid '{oid}': {e}")), - }) - })?; - let timestamp = row.try_read::("timestamp")?; - let timestamp = LocalTime::from_millis(timestamp as u128); - let addresses = self.addresses(&nid)?; - - Ok(types::Seed { - nid, - addresses, - synced_at: SyncedAt { oid, timestamp }, - }) - }))) - } - - fn seeded_by( - &self, - nid: &NodeId, - ) -> Result> + '_>, Error> { - let mut stmt = self.db.prepare( - "SELECT repo, head, timestamp - FROM `repo-sync-status` - WHERE node = ?", - )?; - stmt.bind((1, nid))?; - - Ok(Box::new(stmt.into_iter().map(|row| { - let row = row?; - let rid = row.try_read::("repo")?; - let oid = row.try_read::<&str, _>("head")?; - let oid = Oid::from_str(oid).map_err(|e| { - Error::Internal(sql::Error { - code: None, - message: Some(format!("sql: invalid oid '{oid}': {e}")), - }) - })?; - let timestamp = row.try_read::("timestamp")?; - let timestamp = LocalTime::from_millis(timestamp as u128); - - Ok((rid, SyncedAt { oid, timestamp })) - }))) - } - fn entries(&self) -> Result>, Error> { let mut stmt = self .db @@ -381,7 +268,10 @@ impl Store for Book { } } -impl AliasStore for Book { +impl AliasStore for T +where + T: Store, +{ /// Retrieve `alias` of given node. /// Calls `Self::get` under the hood. fn alias(&self, nid: &NodeId) -> Option { @@ -391,62 +281,6 @@ impl AliasStore for Book { } } -/// Address store. -/// -/// Used to store node addresses and metadata. -pub trait Store { - /// Get the information we have about a node. - fn get(&self, id: &NodeId) -> Result, Error>; - /// Get the addresses of a node. - fn addresses(&self, node: &NodeId) -> Result, Error>; - /// Insert a node with associated addresses into the store. - /// - /// Returns `true` if the node or addresses were updated, and `false` otherwise. - fn insert( - &mut self, - node: &NodeId, - features: node::Features, - alias: Alias, - pow: u32, - timestamp: Timestamp, - addrs: impl IntoIterator, - ) -> Result; - /// Remove a node from the store. - fn remove(&mut self, id: &NodeId) -> Result; - /// Mark a repo as synced on the given node. - fn synced( - &mut self, - rid: &Id, - nid: &NodeId, - at: Oid, - timestamp: Timestamp, - ) -> Result; - /// Get nodes that have synced the given repo. - fn seeds( - &self, - rid: &Id, - ) -> Result> + '_>, Error>; - /// Get the repos seeded by the given node. - fn seeded_by( - &self, - nid: &NodeId, - ) -> Result> + '_>, Error>; - /// Returns the number of addresses. - fn len(&self) -> Result; - /// Returns true if there are no addresses. - fn is_empty(&self) -> Result { - self.len().map(|l| l == 0) - } - /// Get the address entries in the store. - fn entries(&self) -> Result>, Error>; - /// Mark a node as attempted at a certain time. - fn attempted(&self, nid: &NodeId, addr: &Address, time: Timestamp) -> Result<(), Error>; - /// Mark a node as successfully connected at a certain time. - fn connected(&self, nid: &NodeId, addr: &Address, time: Timestamp) -> Result<(), Error>; - /// Mark a node as disconnected. - fn disconnected(&mut self, nid: &NodeId, addr: &Address, transient: bool) -> Result<(), Error>; -} - impl TryFrom<&sql::Value> for Source { type Error = sql::Error; @@ -521,7 +355,7 @@ mod test { fn test_empty() { let tmp = tempfile::tempdir().unwrap(); let path = tmp.path().join("cache"); - let cache = Book::open(path).unwrap(); + let cache = Database::open(path).unwrap(); assert!(cache.is_empty().unwrap()); } @@ -529,7 +363,7 @@ mod test { #[test] fn test_get_none() { let alice = arbitrary::gen::(1); - let cache = Book::memory().unwrap(); + let cache = Database::memory().unwrap(); let result = cache.get(&alice).unwrap(); assert!(result.is_none()); @@ -538,7 +372,7 @@ mod test { #[test] fn test_remove_nothing() { let alice = arbitrary::gen::(1); - let mut cache = Book::memory().unwrap(); + let mut cache = Database::memory().unwrap(); let removed = cache.remove(&alice).unwrap(); assert!(!removed); @@ -547,7 +381,7 @@ mod test { #[test] fn test_alias() { let alice = arbitrary::gen::(1); - let mut cache = Book::memory().unwrap(); + let mut cache = Database::memory().unwrap(); let features = node::Features::SEED; let timestamp = LocalTime::now().as_millis(); @@ -567,7 +401,7 @@ mod test { #[test] fn test_insert_and_get() { let alice = arbitrary::gen::(1); - let mut cache = Book::memory().unwrap(); + let mut cache = Database::memory().unwrap(); let features = node::Features::SEED; let timestamp = LocalTime::now().as_millis(); @@ -602,7 +436,7 @@ mod test { #[test] fn test_insert_duplicate() { let alice = arbitrary::gen::(1); - let mut cache = Book::memory().unwrap(); + let mut cache = Database::memory().unwrap(); let features = node::Features::SEED; let timestamp = LocalTime::now().as_millis(); let alias = Alias::new("alice"); @@ -630,7 +464,7 @@ mod test { #[test] fn test_insert_and_update() { let alice = arbitrary::gen::(1); - let mut cache = Book::memory().unwrap(); + let mut cache = Database::memory().unwrap(); let timestamp = LocalTime::now().as_millis(); let features = node::Features::SEED; let alias1 = Alias::new("alice"); @@ -685,7 +519,7 @@ mod test { fn test_insert_and_remove() { let alice = arbitrary::gen::(1); let bob = arbitrary::gen::(1); - let mut cache = Book::memory().unwrap(); + let mut cache = Database::memory().unwrap(); let timestamp = LocalTime::now().as_millis(); let features = node::Features::SEED; let alice_alias = Alias::new("alice"); @@ -732,7 +566,7 @@ mod test { fn test_entries() { let ids = arbitrary::vec::(16); let mut rng = fastrand::Rng::new(); - let mut cache = Book::memory().unwrap(); + let mut cache = Database::memory().unwrap(); let mut expected = Vec::new(); let timestamp = LocalTime::now().as_millis(); let features = node::Features::SEED; diff --git a/radicle/src/node/address/types.rs b/radicle/src/node/address/types.rs deleted file mode 100644 index 2b9cbcfc..00000000 --- a/radicle/src/node/address/types.rs +++ /dev/null @@ -1,238 +0,0 @@ -use std::cell::RefCell; -use std::hash; -use std::ops::{Deref, DerefMut}; - -use localtime::LocalTime; -use nonempty::NonEmpty; - -use crate::collections::RandomMap; -use crate::git; -use crate::node::{Address, Alias}; -use crate::prelude::{NodeId, Timestamp}; -use crate::storage::{refs::RefsAt, ReadRepository, RemoteId}; -use crate::{node, profile}; - -/// A map with the ability to randomly select values. -#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] -#[serde(transparent)] -pub struct AddressBook { - inner: RandomMap, - #[serde(skip)] - rng: RefCell, -} - -impl AddressBook { - /// Create a new address book. - pub fn new(rng: fastrand::Rng) -> Self { - Self { - inner: RandomMap::with_hasher(rng.clone().into()), - rng: RefCell::new(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.borrow_mut().usize(..pairs.len()); - let pair = pairs[ix]; // Can't fail. - - Some(pair) - } else { - None - } - } - - /// Return a new address book with the given RNG. - pub fn with(self, rng: fastrand::Rng) -> Self { - Self { - inner: self.inner, - rng: RefCell::new(rng), - } - } -} - -impl AddressBook { - /// Return a shuffled iterator. - pub fn shuffled(&self) -> std::vec::IntoIter<(&K, &V)> { - let mut items = self.inner.iter().collect::>(); - items.sort_by_key(|(k, _)| *k); - self.rng.borrow_mut().shuffle(&mut items); - - items.into_iter() - } - - /// Turn this object into a shuffled iterator. - pub fn into_shuffled(self) -> impl Iterator { - let mut items = self.inner.into_iter().collect::>(); - items.sort_by_key(|(k, _)| *k); - self.rng.borrow_mut().shuffle(&mut items); - - items.into_iter() - } - - /// Cycle through the keys at random. The random cycle repeats ad-infintum. - pub fn cycle(&self) -> impl Iterator { - self.shuffled().map(|(k, _)| k).cycle() - } -} - -impl FromIterator<(K, V)> for AddressBook { - fn from_iter>(iter: T) -> Self { - let rng = profile::env::rng(); - let mut inner = RandomMap::with_hasher(rng.clone().into()); - - for (k, v) in iter { - inner.insert(k, v); - } - Self { - inner, - rng: RefCell::new(rng), - } - } -} - -impl Deref for AddressBook { - type Target = RandomMap; - - fn deref(&self) -> &Self::Target { - &self.inner - } -} - -impl DerefMut for AddressBook { - fn deref_mut(&mut self) -> &mut Self::Target { - &mut self.inner - } -} - -/// Node public data. -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct Node { - /// Advertized alias. - pub alias: Alias, - /// Advertized features. - pub features: node::Features, - /// Advertized addresses - pub addrs: Vec, - /// Proof-of-work included in node announcement. - pub pow: u32, - /// When this data was published. - pub timestamp: Timestamp, -} - -/// A known address. -#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] -#[serde(rename_all = "camelCase")] -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. - #[serde(with = "crate::serde_ext::localtime::option::time")] - pub last_success: Option, - /// Last time this address was tried. - #[serde(with = "crate::serde_ext::localtime::option::time")] - pub last_attempt: Option, - /// Whether this address has been banned. - pub banned: bool, -} - -impl KnownAddress { - /// Create a new known address. - pub fn new(addr: Address, source: Source) -> Self { - Self { - addr, - source, - last_success: None, - last_attempt: None, - banned: false, - } - } -} - -/// Address source. Specifies where an address originated from. -#[derive(Debug, Copy, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] -#[serde(rename_all = "camelCase")] -pub enum Source { - /// An address that was shared by another peer. - Peer, - /// An bootstrap node address. - Bootstrap, - /// 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::Bootstrap => write!(f, "Bootstrap"), - Self::Imported => write!(f, "Imported"), - } - } -} - -/// Holds an oid and timestamp. -#[derive(Debug, Copy, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] -#[serde(rename_all = "camelCase")] -pub struct SyncedAt { - /// Head of `rad/sigrefs`. - pub oid: git_ext::Oid, - /// When these refs were synced. - #[serde(with = "crate::serde_ext::localtime::time")] - pub timestamp: LocalTime, -} - -impl SyncedAt { - /// Load a new [`SyncedAt`] for the given remote. - pub fn load(repo: &S, remote: RemoteId) -> Result { - let refs = RefsAt::new(repo, remote)?; - let oid = refs.at; - - Self::new(oid, repo) - } - - /// Create a new [`SyncedAt`] given an OID, by looking up the timestamp in the repo. - pub fn new(oid: git::ext::Oid, repo: &S) -> Result { - let timestamp = repo.commit(oid)?.time(); - let timestamp = LocalTime::from_secs(timestamp.seconds() as u64); - - Ok(Self { oid, timestamp }) - } -} - -impl Ord for SyncedAt { - fn cmp(&self, other: &Self) -> std::cmp::Ordering { - self.timestamp.cmp(&other.timestamp) - } -} - -impl PartialOrd for SyncedAt { - fn partial_cmp(&self, other: &Self) -> Option { - Some(self.cmp(other)) - } -} - -/// Seed of a specific repository. -#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] -#[serde(rename_all = "camelCase")] -pub struct Seed { - /// The Node ID. - pub nid: NodeId, - /// Known addresses for this node. - pub addresses: Vec, - /// Sync information for a given repo. - pub synced_at: SyncedAt, -} diff --git a/radicle/src/node/db.rs b/radicle/src/node/db.rs new file mode 100644 index 00000000..bfc5e522 --- /dev/null +++ b/radicle/src/node/db.rs @@ -0,0 +1,77 @@ +use std::path::Path; +use std::{fmt, time}; + +use sqlite as sql; +use thiserror::Error; + +/// How long to wait for the database lock to be released before failing a read. +const DB_READ_TIMEOUT: time::Duration = time::Duration::from_secs(3); +/// How long to wait for the database lock to be released before failing a write. +const DB_WRITE_TIMEOUT: time::Duration = time::Duration::from_secs(6); + +#[derive(Error, Debug)] +pub enum Error { + /// An Internal error. + #[error("internal error: {0}")] + Internal(#[from] sql::Error), +} + +/// A file-backed database storing information about the network. +pub struct Database { + pub db: sql::Connection, +} + +impl fmt::Debug for Database { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("Database").finish() + } +} + +impl From for Database { + fn from(db: sql::Connection) -> Self { + Self { db } + } +} + +impl Database { + const SCHEMA: &'static str = include_str!("db/schema.sql"); + const PRAGMA: &'static str = "PRAGMA foreign_keys = ON"; + + /// 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 mut db = sql::Connection::open_with_flags( + path, + sqlite::OpenFlags::new() + .with_create() + .with_read_write() + .with_full_mutex(), + )?; + db.set_busy_timeout(DB_WRITE_TIMEOUT.as_millis() as usize)?; + db.execute(Self::PRAGMA)?; + db.execute(Self::SCHEMA)?; + + Ok(Self { db }) + } + + /// Same as [`Self::open`], but in read-only mode. This is useful to have multiple + /// open databases, as no locking is required. + pub fn reader>(path: P) -> Result { + let mut db = + sql::Connection::open_with_flags(path, sqlite::OpenFlags::new().with_read_only())?; + db.set_busy_timeout(DB_READ_TIMEOUT.as_millis() as usize)?; + db.execute(Self::PRAGMA)?; + db.execute(Self::SCHEMA)?; + + Ok(Self { db }) + } + + /// Create a new in-memory address book. + pub fn memory() -> Result { + let db = sql::Connection::open(":memory:")?; + db.execute(Self::PRAGMA)?; + db.execute(Self::SCHEMA)?; + + Ok(Self { db }) + } +} diff --git a/radicle/src/node/address/schema.sql b/radicle/src/node/db/schema.sql similarity index 83% rename from radicle/src/node/address/schema.sql rename to radicle/src/node/db/schema.sql index 6cb07291..2655eb84 100644 --- a/radicle/src/node/address/schema.sql +++ b/radicle/src/node/db/schema.sql @@ -1,6 +1,4 @@ --- --- Address book SQL schema. --- +-- Discovered nodes. create table if not exists "nodes" ( -- Node ID. "id" text primary key not null, @@ -17,6 +15,7 @@ create table if not exists "nodes" ( -- ) strict; +-- Node addresses. create table if not exists "addresses" ( -- Node ID. "node" text not null references "nodes" ("id") on delete cascade, @@ -42,9 +41,24 @@ create table if not exists "addresses" ( -- ) strict; +-- Routing table. Tracks inventories. +create table if not exists "routing" ( + -- Repository being seeded. + "repo" text not null, + -- Node ID. + -- TODO: Add foreign-key constraint. + "node" text not null, + -- UNIX time at which this entry was added or refreshed. + "timestamp" integer not null, + + primary key ("repo", "node") +); + +-- Gossip message store. create table if not exists "announcements" ( -- Node ID. - "node" text not null references "nodes" ("id") on delete cascade, + -- TODO: Add foreign-key constraint. + "node" text not null, -- Repo ID, if any, for example in ref announcements. -- For other announcement types, this should be an empty string. "repo" text not null, @@ -64,6 +78,7 @@ create table if not exists "announcements" ( "timestamp" integer not null, -- unique ("node", "repo", "type") + -- ) strict; -- Repository sync status. @@ -78,4 +93,5 @@ create table if not exists "repo-sync-status" ( "timestamp" integer not null, -- unique ("repo", "node") + -- ) strict; diff --git a/radicle/src/node/routing.rs b/radicle/src/node/routing.rs index a47dc898..1324c3ca 100644 --- a/radicle/src/node/routing.rs +++ b/radicle/src/node/routing.rs @@ -1,21 +1,15 @@ use std::collections::HashSet; -use std::path::Path; -use std::{fmt, time}; use sqlite as sql; use thiserror::Error; +use crate::node::Database; use crate::{ prelude::Timestamp, prelude::{Id, NodeId}, sql::transaction, }; -/// How long to wait for the database lock to be released before failing a read. -const DB_READ_TIMEOUT: time::Duration = time::Duration::from_secs(3); -/// How long to wait for the database lock to be released before failing a write. -const DB_WRITE_TIMEOUT: time::Duration = time::Duration::from_secs(6); - /// Result of inserting into the routing table. #[derive(Debug, Copy, Clone, PartialEq, Eq)] pub enum InsertResult { @@ -38,50 +32,6 @@ pub enum Error { UnitOverflow, } -/// Persistent file storage for a routing table. -pub struct Table { - db: sql::Connection, -} - -impl fmt::Debug for Table { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - write!(f, "Table(..)") - } -} - -impl Table { - const SCHEMA: &'static str = include_str!("routing/schema.sql"); - - /// Open a routing file store at the given path. Creates a new empty store - /// if an existing store isn't found. - pub fn open>(path: P) -> Result { - let mut db = sql::Connection::open(path)?; - db.set_busy_timeout(DB_WRITE_TIMEOUT.as_millis() as usize)?; - db.execute(Self::SCHEMA)?; - - Ok(Self { db }) - } - - /// Same as [`Self::open`], but in read-only mode. This is useful to have multiple - /// open databases, as no locking is required. - pub fn reader>(path: P) -> Result { - let mut db = - sql::Connection::open_with_flags(path, sqlite::OpenFlags::new().with_read_only())?; - db.set_busy_timeout(DB_READ_TIMEOUT.as_millis() as usize)?; - db.execute(Self::SCHEMA)?; - - Ok(Self { db }) - } - - /// Create a new in-memory routing table. - pub fn memory() -> Result { - let db = sql::Connection::open(":memory:")?; - db.execute(Self::SCHEMA)?; - - Ok(Self { db }) - } -} - /// Backing store for a routing table. pub trait Store { /// Get the nodes seeding the given id. @@ -113,11 +63,11 @@ pub trait Store { fn count(&self, id: &Id) -> Result; } -impl Store for Table { +impl Store for Database { fn get(&self, id: &Id) -> Result, Error> { let mut stmt = self .db - .prepare("SELECT (node) FROM routing WHERE resource = ?")?; + .prepare("SELECT (node) FROM routing WHERE repo = ?")?; stmt.bind((1, id))?; let mut nodes = HashSet::new(); @@ -128,14 +78,12 @@ impl Store for Table { } fn get_resources(&self, node: &NodeId) -> Result, Error> { - let mut stmt = self - .db - .prepare("SELECT resource FROM routing WHERE node = ?")?; + let mut stmt = self.db.prepare("SELECT repo FROM routing WHERE node = ?")?; stmt.bind((1, node))?; let mut resources = HashSet::new(); for row in stmt.into_iter() { - resources.insert(row?.read::("resource")); + resources.insert(row?.read::("repo")); } Ok(resources) } @@ -143,13 +91,13 @@ impl Store for Table { fn entry(&self, id: &Id, node: &NodeId) -> Result, Error> { let mut stmt = self .db - .prepare("SELECT (time) FROM routing WHERE resource = ? AND node = ?")?; + .prepare("SELECT (timestamp) FROM routing WHERE repo = ? AND node = ?")?; stmt.bind((1, id))?; stmt.bind((2, node))?; if let Some(Ok(row)) = stmt.into_iter().next() { - return Ok(Some(row.read::("time") as Timestamp)); + return Ok(Some(row.read::("timestamp") as Timestamp)); } Ok(None) } @@ -166,18 +114,18 @@ impl Store for Table { transaction(&self.db, |db| { for id in ids.into_iter() { let mut stmt = - db.prepare("SELECT (time) FROM routing WHERE resource = ? AND node = ?")?; + db.prepare("SELECT (timestamp) FROM routing WHERE repo = ? AND node = ?")?; stmt.bind((1, id))?; stmt.bind((2, &node))?; let existed = stmt.into_iter().next().is_some(); let mut stmt = db.prepare( - "INSERT INTO routing (resource, node, time) + "INSERT INTO routing (repo, node, timestamp) VALUES (?, ?, ?) ON CONFLICT DO UPDATE - SET time = ?3 - WHERE time < ?3", + SET timestamp = ?3 + WHERE timestamp < ?3", )?; stmt.bind((1, id))?; @@ -200,12 +148,12 @@ impl Store for Table { fn entries(&self) -> Result>, Error> { let mut stmt = self .db - .prepare("SELECT resource, node FROM routing ORDER BY resource")? + .prepare("SELECT repo, node FROM routing ORDER BY repo")? .into_iter(); let mut entries = Vec::new(); while let Some(Ok(row)) = stmt.next() { - let id = row.read("resource"); + let id = row.read("repo"); let node = row.read("node"); entries.push((id, node)); @@ -216,7 +164,7 @@ impl Store for Table { fn remove(&mut self, id: &Id, node: &NodeId) -> Result { let mut stmt = self .db - .prepare("DELETE FROM routing WHERE resource = ? AND node = ?")?; + .prepare("DELETE FROM routing WHERE repo = ? AND node = ?")?; stmt.bind((1, id))?; stmt.bind((2, node))?; @@ -245,7 +193,7 @@ impl Store for Table { let mut stmt = self.db.prepare( "DELETE FROM routing WHERE rowid IN - (SELECT rowid FROM routing WHERE time < ? LIMIT ?)", + (SELECT rowid FROM routing WHERE timestamp < ? LIMIT ?)", )?; stmt.bind((1, oldest))?; stmt.bind((2, limit))?; @@ -257,7 +205,7 @@ impl Store for Table { fn count(&self, id: &Id) -> Result { let mut stmt = self .db - .prepare("SELECT COUNT(*) FROM routing WHERE resource = ?")?; + .prepare("SELECT COUNT(*) FROM routing WHERE repo = ?")?; stmt.bind((1, id))?; @@ -284,7 +232,7 @@ mod test { fn test_insert_and_get() { let ids = arbitrary::set::(5..10); let nodes = arbitrary::set::(5..10); - let mut db = Table::open(":memory:").unwrap(); + let mut db = Database::open(":memory:").unwrap(); for node in &nodes { assert_eq!( @@ -307,7 +255,7 @@ mod test { fn test_insert_and_get_resources() { let ids = arbitrary::set::(5..10); let nodes = arbitrary::set::(5..10); - let mut db = Table::open(":memory:").unwrap(); + let mut db = Database::open(":memory:").unwrap(); for node in &nodes { db.insert(&ids, *node, 0).unwrap(); @@ -325,7 +273,7 @@ mod test { fn test_entries() { let ids = arbitrary::set::(6..9); let nodes = arbitrary::set::(6..9); - let mut db = Table::open(":memory:").unwrap(); + let mut db = Database::open(":memory:").unwrap(); for node in &nodes { assert!(db @@ -348,7 +296,7 @@ mod test { fn test_insert_and_remove() { let ids = arbitrary::set::(5..10); let nodes = arbitrary::set::(5..10); - let mut db = Table::open(":memory:").unwrap(); + let mut db = Database::open(":memory:").unwrap(); for node in &nodes { db.insert(&ids, *node, 0).unwrap(); @@ -367,7 +315,7 @@ mod test { fn test_insert_duplicate() { let id = arbitrary::gen::(1); let node = arbitrary::gen::(1); - let mut db = Table::open(":memory:").unwrap(); + let mut db = Database::open(":memory:").unwrap(); assert_eq!( db.insert([&id], node, 0).unwrap(), @@ -387,7 +335,7 @@ mod test { fn test_insert_existing_updated_time() { let id = arbitrary::gen::(1); let node = arbitrary::gen::(1); - let mut db = Table::open(":memory:").unwrap(); + let mut db = Database::open(":memory:").unwrap(); assert_eq!( db.insert([&id], node, 0).unwrap(), @@ -405,7 +353,7 @@ mod test { let id1 = arbitrary::gen::(1); let id2 = arbitrary::gen::(1); let node = arbitrary::gen::(1); - let mut db = Table::open(":memory:").unwrap(); + let mut db = Database::open(":memory:").unwrap(); assert_eq!( db.insert([&id1], node, 0).unwrap(), @@ -431,7 +379,7 @@ mod test { fn test_remove_redundant() { let id = arbitrary::gen::(1); let node = arbitrary::gen::(1); - let mut db = Table::open(":memory:").unwrap(); + let mut db = Database::open(":memory:").unwrap(); assert_eq!( db.insert([&id], node, 0).unwrap(), @@ -443,7 +391,7 @@ mod test { #[test] fn test_len() { - let mut db = Table::open(":memory:").unwrap(); + let mut db = Database::open(":memory:").unwrap(); let ids = arbitrary::vec::(10); let node = arbitrary::gen(1); @@ -458,7 +406,7 @@ mod test { let now = LocalTime::now(); let ids = arbitrary::vec::(10); let nodes = arbitrary::vec::(10); - let mut db = Table::open(":memory:").unwrap(); + let mut db = Database::open(":memory:").unwrap(); for node in &nodes { let time = rng.u64(..now.as_millis()); @@ -488,7 +436,7 @@ mod test { fn test_count() { let id = arbitrary::gen::(1); let nodes = arbitrary::set::(5..10); - let mut db = Table::open(":memory:").unwrap(); + let mut db = Database::open(":memory:").unwrap(); for node in &nodes { db.insert([&id], *node, 0).unwrap(); diff --git a/radicle/src/node/routing/schema.sql b/radicle/src/node/routing/schema.sql deleted file mode 100644 index 7f600dff..00000000 --- a/radicle/src/node/routing/schema.sql +++ /dev/null @@ -1,13 +0,0 @@ --- --- Routing table SQL schema. --- -create table if not exists "routing" ( - -- Resource being seeded. - "resource" text not null, - -- Node ID. - "node" text not null, - -- UNIX time at which this entry was added or refreshed. - "time" integer not null, - - primary key ("resource", "node") -); diff --git a/radicle/src/node/seed.rs b/radicle/src/node/seed.rs new file mode 100644 index 00000000..3408e196 --- /dev/null +++ b/radicle/src/node/seed.rs @@ -0,0 +1,62 @@ +pub mod store; +pub use store::{Error, Store}; + +use localtime::LocalTime; + +use crate::git; +use crate::node::KnownAddress; +use crate::prelude::NodeId; +use crate::storage::{refs::RefsAt, ReadRepository, RemoteId}; + +/// Holds an oid and timestamp. +#[derive(Debug, Copy, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct SyncedAt { + /// Head of `rad/sigrefs`. + pub oid: git_ext::Oid, + /// When these refs were synced. + #[serde(with = "crate::serde_ext::localtime::time")] + pub timestamp: LocalTime, +} + +impl SyncedAt { + /// Load a new [`SyncedAt`] for the given remote. + pub fn load(repo: &S, remote: RemoteId) -> Result { + let refs = RefsAt::new(repo, remote)?; + let oid = refs.at; + + Self::new(oid, repo) + } + + /// Create a new [`SyncedAt`] given an OID, by looking up the timestamp in the repo. + pub fn new(oid: git::ext::Oid, repo: &S) -> Result { + let timestamp = repo.commit(oid)?.time(); + let timestamp = LocalTime::from_secs(timestamp.seconds() as u64); + + Ok(Self { oid, timestamp }) + } +} + +impl Ord for SyncedAt { + fn cmp(&self, other: &Self) -> std::cmp::Ordering { + self.timestamp.cmp(&other.timestamp) + } +} + +impl PartialOrd for SyncedAt { + fn partial_cmp(&self, other: &Self) -> Option { + Some(self.cmp(other)) + } +} + +/// Seed of a specific repository that has been synced at least once. +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct SyncedSeed { + /// The Node ID. + pub nid: NodeId, + /// Known addresses for this node. + pub addresses: Vec, + /// Sync information for a given repo. + pub synced_at: SyncedAt, +} diff --git a/radicle/src/node/seed/store.rs b/radicle/src/node/seed/store.rs new file mode 100644 index 00000000..08da141c --- /dev/null +++ b/radicle/src/node/seed/store.rs @@ -0,0 +1,133 @@ +#![allow(clippy::type_complexity)] +use std::str::FromStr; + +use localtime::LocalTime; +use sqlite as sql; +use thiserror::Error; + +use crate::git::Oid; +use crate::node::address; +use crate::node::address::Store as _; +use crate::node::NodeId; +use crate::node::{seed::SyncedSeed, Database, SyncedAt}; +use crate::prelude::{Id, Timestamp}; + +#[derive(Error, Debug)] +pub enum Error { + /// An Internal error. + #[error("internal error: {0}")] + Internal(#[from] sql::Error), + /// An address store error. + #[error("address store error: {0}")] + Addresses(#[from] address::Error), +} + +/// Seed store. +/// +/// Used to store seed sync statuses. +pub trait Store: address::Store { + /// Mark a repo as synced on the given node. + fn synced( + &mut self, + rid: &Id, + nid: &NodeId, + at: Oid, + timestamp: Timestamp, + ) -> Result; + /// Get the repos seeded by the given node. + fn seeded_by( + &self, + nid: &NodeId, + ) -> Result> + '_>, Error>; + /// Get nodes that have synced the given repo. + fn seeds_for( + &self, + rid: &Id, + ) -> Result> + '_>, Error>; +} + +impl Store for Database { + fn synced( + &mut self, + rid: &Id, + nid: &NodeId, + at: Oid, + timestamp: Timestamp, + ) -> Result { + let mut stmt = self.db.prepare( + "INSERT INTO `repo-sync-status` (repo, node, head, timestamp) + VALUES (?1, ?2, ?3, ?4) + ON CONFLICT DO UPDATE + SET head = ?3, timestamp = ?4 + WHERE timestamp < ?4", + )?; + stmt.bind((1, rid))?; + stmt.bind((2, nid))?; + stmt.bind((3, at.to_string().as_str()))?; + stmt.bind((4, timestamp as i64))?; + stmt.next()?; + + Ok(self.db.change_count() > 0) + } + + fn seeds_for( + &self, + rid: &Id, + ) -> Result> + '_>, Error> { + let mut stmt = self.db.prepare( + "SELECT node, head, timestamp + FROM `repo-sync-status` + WHERE repo = ?", + )?; + stmt.bind((1, rid))?; + + Ok(Box::new(stmt.into_iter().map(|row| { + let row = row?; + let nid = row.try_read::("node")?; + let oid = row.try_read::<&str, _>("head")?; + let oid = Oid::from_str(oid).map_err(|e| { + Error::Internal(sql::Error { + code: None, + message: Some(format!("sql: invalid oid '{oid}': {e}")), + }) + })?; + let timestamp = row.try_read::("timestamp")?; + let timestamp = LocalTime::from_millis(timestamp as u128); + let addresses = self.addresses_of(&nid)?; + + Ok(SyncedSeed { + nid, + addresses, + synced_at: SyncedAt { oid, timestamp }, + }) + }))) + } + + fn seeded_by( + &self, + nid: &NodeId, + ) -> Result> + '_>, Error> { + let mut stmt = self.db.prepare( + "SELECT repo, head, timestamp + FROM `repo-sync-status` + WHERE node = ?", + )?; + stmt.bind((1, nid))?; + + Ok(Box::new(stmt.into_iter().map(|row| { + let row = row?; + let rid = row.try_read::("repo")?; + let oid = row.try_read::<&str, _>("head")?; + let oid = Oid::from_str(oid).map_err(|e| { + Error::Internal(sql::Error { + code: None, + message: Some(format!("sql: invalid oid '{oid}': {e}")), + }) + })?; + let timestamp = row.try_read::("timestamp")?; + let timestamp = LocalTime::from_millis(timestamp as u128); + + Ok((rid, SyncedAt { oid, timestamp })) + }))) + } +} diff --git a/radicle/src/profile.rs b/radicle/src/profile.rs index 3a3c3730..73b81454 100644 --- a/radicle/src/profile.rs +++ b/radicle/src/profile.rs @@ -20,7 +20,7 @@ use thiserror::Error; use crate::crypto::ssh::agent::Agent; use crate::crypto::ssh::{keystore, Keystore, Passphrase}; use crate::crypto::{PublicKey, Signer}; -use crate::node::{address, routing, tracking, Alias, AliasStore}; +use crate::node::{tracking, Alias, AliasStore}; use crate::prelude::Did; use crate::prelude::{Id, NodeId}; use crate::storage::git::transport; @@ -148,7 +148,7 @@ pub enum Error { #[error(transparent)] TrackingStore(#[from] node::tracking::store::Error), #[error(transparent)] - AddressStore(#[from] node::address::Error), + DatabaseStore(#[from] node::db::Error), } #[derive(Debug, Error)] @@ -357,22 +357,6 @@ impl Profile { Ok(config) } - /// Return a read-only handle to the routing database of the node. - pub fn routing(&self) -> Result { - let path = self.home.node().join(node::ROUTING_DB_FILE); - let router = routing::Table::reader(path)?; - - Ok(router) - } - - /// Return a handle to the addresses database of the node. - pub fn addresses(&self) -> Result { - let path = self.home.node().join(node::ADDRESS_DB_FILE); - let addresses = address::Book::reader(path)?; - - Ok(addresses) - } - /// Get radicle home. pub fn home(&self) -> &Home { &self.home @@ -381,12 +365,9 @@ impl Profile { /// Return a multi-source store for aliases. pub fn aliases(&self) -> Aliases { let tracking = self.home.tracking().ok(); - let addresses = self.home.addresses().ok(); + let db = self.home.database().ok(); - Aliases { - tracking, - addresses, - } + Aliases { tracking, db } } } @@ -408,7 +389,7 @@ impl std::ops::DerefMut for Profile { /// them one by one when asking for an alias. pub struct Aliases { tracking: Option, - addresses: Option, + db: Option, } impl AliasStore for Aliases { @@ -418,7 +399,7 @@ impl AliasStore for Aliases { self.tracking .as_ref() .and_then(|db| db.alias(nid)) - .or_else(|| self.addresses.as_ref().and_then(|db| db.alias(nid))) + .or_else(|| self.db.as_ref().and_then(|db| db.alias(nid))) } } @@ -522,36 +503,20 @@ impl Home { Ok(config) } - /// Return a read-only handle to the routing database of the node. - pub fn routing(&self) -> Result { - let path = self.node().join(node::ROUTING_DB_FILE); - let router = routing::Table::reader(path)?; + /// Return a handle to a read-only database of the node. + pub fn database(&self) -> Result { + let path = self.node().join(node::NODE_DB_FILE); + let db = node::Database::reader(path)?; - Ok(router) + Ok(db) } - /// Return a read-write handle to the routing database of the node. - pub fn routing_mut(&self) -> Result { - let path = self.node().join(node::ROUTING_DB_FILE); - let router = routing::Table::open(path)?; + /// Return a handle to the database of the node. + pub fn database_mut(&self) -> Result { + let path = self.node().join(node::NODE_DB_FILE); + let db = node::Database::open(path)?; - Ok(router) - } - - /// Return a handle to a read-only addresses database of the node. - pub fn addresses(&self) -> Result { - let path = self.node().join(node::ADDRESS_DB_FILE); - let addresses = address::Book::reader(path)?; - - Ok(addresses) - } - - /// Return a handle to the addresses database of the node. - pub fn addresses_mut(&self) -> Result { - let path = self.node().join(node::ADDRESS_DB_FILE); - let addresses = address::Book::open(path)?; - - Ok(addresses) + Ok(db) } }