From 85ac8a1d4db5e0b4337e863485de94124858fa83 Mon Sep 17 00:00:00 2001 From: Alexis Sellier Date: Tue, 4 Oct 2022 14:08:51 +0200 Subject: [PATCH] Implement routing table persistence Uses SQLite for persistence. Signed-off-by: Alexis Sellier --- Cargo.lock | 68 +++++++ radicle-node/Cargo.toml | 2 + radicle-node/src/client.rs | 30 ++- radicle-node/src/client/handle.rs | 10 +- radicle-node/src/control.rs | 9 +- radicle-node/src/service.rs | 76 ++++---- radicle-node/src/service/routing.rs | 196 ++++++++++++++++++++ radicle-node/src/service/routing/schema.sql | 15 ++ radicle-node/src/test/handle.rs | 2 +- radicle-node/src/test/peer.rs | 7 +- radicle-node/src/test/tests.rs | 4 +- radicle-node/src/transport.rs | 20 +- radicle-node/src/wire.rs | 21 ++- radicle/Cargo.toml | 2 + radicle/src/lib.rs | 2 + radicle/src/sql.rs | 48 +++++ 16 files changed, 436 insertions(+), 76 deletions(-) create mode 100644 radicle-node/src/service/routing.rs create mode 100644 radicle-node/src/service/routing/schema.sql create mode 100644 radicle/src/sql.rs diff --git a/Cargo.lock b/Cargo.lock index f3317f47..a3d2b3cf 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2,6 +2,17 @@ # It is not intended for manual editing. version = 3 +[[package]] +name = "ahash" +version = "0.7.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fcb51a0695d8f838b1ee009b3fbf66bda078cd64590202a864a8f3e8c4315c47" +dependencies = [ + "getrandom", + "once_cell", + "version_check", +] + [[package]] name = "android_system_properties" version = "0.1.5" @@ -295,6 +306,18 @@ dependencies = [ "getrandom", ] +[[package]] +name = "fallible-iterator" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4443176a9f2c162692bd3d352d745ef9413eec5782a80d8fd6f8a1ac692a07f7" + +[[package]] +name = "fallible-streaming-iterator" +version = "0.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7360491ce676a36bf9bb3c56c1aa791658183a54d2744120f27285738d90465a" + [[package]] name = "fastrand" version = "1.8.0" @@ -396,6 +419,18 @@ name = "hashbrown" version = "0.12.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8a9ee70c43aaf417c914396645a0fa852624801b24ebb7ae78fe8272889ac888" +dependencies = [ + "ahash", +] + +[[package]] +name = "hashlink" +version = "0.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "69fe1fcf8b4278d860ad0548329f892a3631fb63f82574df68275f34cdbe0ffa" +dependencies = [ + "hashbrown", +] [[package]] name = "hermit-abi" @@ -520,6 +555,17 @@ dependencies = [ "pkg-config", ] +[[package]] +name = "libsqlite3-sys" +version = "0.25.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9f0455f2c1bc9a7caa792907026e469c1d91761fb0ea37cbb16427c77280cf35" +dependencies = [ + "cc", + "pkg-config", + "vcpkg", +] + [[package]] name = "libssh2-sys" version = "0.2.23" @@ -791,6 +837,7 @@ dependencies = [ "quickcheck_macros", "radicle-git-ext", "radicle-ssh", + "rusqlite", "serde", "serde_json", "sha2 0.10.6", @@ -834,6 +881,7 @@ dependencies = [ "quickcheck", "quickcheck_macros", "radicle", + "rusqlite", "serde", "serde_json", "tempfile", @@ -900,6 +948,20 @@ dependencies = [ "winapi", ] +[[package]] +name = "rusqlite" +version = "0.28.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "01e213bc3ecb39ac32e81e51ebe31fd888a940515173e3a18a35f8c6e896422a" +dependencies = [ + "bitflags", + "fallible-iterator", + "fallible-streaming-iterator", + "hashlink", + "libsqlite3-sys", + "smallvec", +] + [[package]] name = "ryu" version = "1.0.11" @@ -993,6 +1055,12 @@ version = "0.3.10" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7bd3e3206899af3f8b12af284fafc038cc1dc2b41d1b89dd17297221c5d225de" +[[package]] +name = "smallvec" +version = "1.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a507befe795404456341dfab10cef66ead4c041f62b8b11bbb92bffe5d0953e0" + [[package]] name = "socket2" version = "0.4.7" diff --git a/radicle-node/Cargo.toml b/radicle-node/Cargo.toml index dfa9aaa1..1b2735a1 100644 --- a/radicle-node/Cargo.toml +++ b/radicle-node/Cargo.toml @@ -19,6 +19,7 @@ log = { version = "0.4.17", features = ["std"] } nakamoto-net = { version = "0.3.0" } nakamoto-net-poll = { version = "0.3.0" } nonempty = { version = "0.8.0", features = ["serialize"] } +rusqlite = { version = "0.28.0", features = ["bundled"] } serde = { version = "1", features = ["derive"] } serde_json = { version = "1", features = ["preserve_order"] } tempfile = { version = "3.3.0" } @@ -27,6 +28,7 @@ thiserror = { version = "1" } [dependencies.radicle] path = "../radicle" version = "0.2.0" +features = ["sql"] [dev-dependencies] radicle = { path = "../radicle", version = "*", features = ["test"] } diff --git a/radicle-node/src/client.rs b/radicle-node/src/client.rs index 753ef675..7e3d03f8 100644 --- a/radicle-node/src/client.rs +++ b/radicle-node/src/client.rs @@ -1,17 +1,38 @@ -use std::net; +use std::{io, net}; use crossbeam_channel as chan; use nakamoto_net::{LocalTime, Reactor}; +use thiserror::Error; use crate::clock::RefClock; use crate::collections::HashMap; use crate::profile::Profile; use crate::service; +use crate::service::routing; use crate::transport::Transport; use crate::wire::Wire; pub mod handle; +/// Directory in `$RAD_HOME` under which node-specific files are stored. +pub const NODE_DIR: &str = "node"; +/// Filename of routing table database under [`NODE_DIR`]. +pub const ROUTING_DB_FILE: &str = "routing.db"; + +/// A client error. +#[derive(Error, Debug)] +pub enum Error { + /// A routing database error. + #[error("routing database error: {0}")] + Routing(#[from] routing::Error), + /// An I/O error. + #[error("i/o error: {0}")] + Io(#[from] io::Error), + /// A networking error. + #[error("network error: {0}")] + Net(#[from] nakamoto_net::error::Error), +} + /// Client configuration. #[derive(Debug, Clone)] pub struct Config { @@ -55,7 +76,7 @@ pub struct Client { } impl Client { - pub fn new(profile: Profile) -> Result { + pub fn new(profile: Profile) -> Result { let (handle, commands) = chan::unbounded::(); let (shutdown, shutdown_recv) = chan::bounded(1); let (listening_send, listening) = chan::bounded(1); @@ -73,24 +94,27 @@ impl Client { }) } - pub fn run(mut self, config: Config) -> Result<(), nakamoto_net::error::Error> { + pub fn run(mut self, config: Config) -> Result<(), Error> { let network = config.service.network; let rng = fastrand::Rng::new(); let time = LocalTime::now(); let storage = self.profile.storage; let signer = self.profile.signer; let addresses = HashMap::with_hasher(rng.clone().into()); + let routing = routing::Table::open(self.profile.home.join(NODE_DIR).join(ROUTING_DB_FILE))?; log::info!("Initializing client ({:?})..", network); let service = service::Service::new( config.service, RefClock::from(time), + routing, storage, addresses, signer, rng, ); + self.reactor.run( &config.listen, Transport::new(Wire::new(service)), diff --git a/radicle-node/src/client/handle.rs b/radicle-node/src/client/handle.rs index 4480a325..12280ed8 100644 --- a/radicle-node/src/client/handle.rs +++ b/radicle-node/src/client/handle.rs @@ -90,11 +90,11 @@ impl traits::Handle for Handle { Ok(()) } - fn routing(&self) -> Result)>, Error> { + fn routing(&self) -> Result, Error> { let (sender, receiver) = chan::unbounded(); let query: Arc = Arc::new(move |state| { - for (id, nodes) in state.routing().iter() { - if sender.send((*id, nodes.iter().cloned().collect())).is_err() { + for (id, node) in state.routing().entries()? { + if sender.send((id, node)).is_err() { break; } } @@ -155,8 +155,8 @@ pub mod traits { fn command(&self, cmd: service::Command) -> Result<(), Error>; /// Ask the client to shutdown. fn shutdown(self) -> Result<(), Error>; - /// Query the routing table. - fn routing(&self) -> Result)>, Error>; + /// Query the routing table entries. + fn routing(&self) -> Result, Error>; /// Query the peer session state. fn sessions(&self) -> Result, Error>; /// Query the inventory. diff --git a/radicle-node/src/control.rs b/radicle-node/src/control.rs index 388e123d..0c5b09b5 100644 --- a/radicle-node/src/control.rs +++ b/radicle-node/src/control.rs @@ -106,13 +106,8 @@ fn drain(stream: &UnixStream, handle: &H) -> Result<(), DrainError> { Ok(c) => { let mut writer = LineWriter::new(stream); - for (id, seeds) in c.iter() { - let seeds = seeds - .into_iter() - .map(String::from) - .collect::>() - .join(" "); - writeln!(writer, "{id} {seeds}",)?; + for (id, seed) in c.iter() { + writeln!(writer, "{id} {seed}",)?; } } Err(e) => return Err(DrainError::Client(e)), diff --git a/radicle-node/src/service.rs b/radicle-node/src/service.rs index a94635ce..10fefcc9 100644 --- a/radicle-node/src/service.rs +++ b/radicle-node/src/service.rs @@ -3,6 +3,7 @@ pub mod filter; pub mod message; pub mod peer; pub mod reactor; +pub mod routing; use std::collections::BTreeMap; use std::ops::{Deref, DerefMut}; @@ -22,7 +23,6 @@ use crate::address_book; use crate::address_book::AddressBook; use crate::address_manager::AddressManager; use crate::clock::{RefClock, Timestamp}; -use crate::collections::{HashMap, HashSet}; use crate::crypto; use crate::crypto::{Signer, Verified}; use crate::git; @@ -55,8 +55,6 @@ pub const MAX_TIME_DELTA: LocalDuration = LocalDuration::from_mins(60); /// Network node identifier. pub type NodeId = crypto::PublicKey; -/// Network routing table. Keeps track of where projects are hosted. -pub type Routing = HashMap>; /// A service event. #[derive(Debug, Clone)] @@ -148,18 +146,20 @@ impl fmt::Debug for Command { pub enum CommandError { #[error(transparent)] Storage(#[from] storage::Error), + #[error(transparent)] + Routing(#[from] routing::Error), } #[derive(Debug)] -pub struct Service { +pub struct Service { /// Service configuration. config: Config, /// Our cryptographic signer and key. signer: G, /// Project storage. storage: S, - /// Tracks the location of projects. - routing: Routing, + /// Network routing table. Keeps track of where projects are located. + routing: R, /// State relating to gossip. gossip: Gossip, /// Peer sessions, currently or recently connected. @@ -188,7 +188,7 @@ pub struct Service { start_time: LocalTime, } -impl Service +impl Service where G: crypto::Signer, { @@ -203,8 +203,9 @@ where } } -impl Service +impl Service where + R: routing::Store, A: address_book::Store, S: WriteStorage + 'static, G: crypto::Signer, @@ -212,13 +213,13 @@ where pub fn new( config: Config, clock: RefClock, + routing: R, storage: S, addresses: A, signer: G, rng: Rng, ) -> Self { let addrmgr = AddressManager::new(addresses); - let routing = HashMap::with_hasher(rng.clone().into()); let sessions = Sessions::new(rng.clone()); let network = config.network; @@ -243,15 +244,14 @@ where } } - pub fn seeds(&self, id: &Id) -> Box + '_> { - if let Some(peers) = self.routing.get(id) { - Box::new( - peers - .iter() - .filter_map(|id| self.sessions.by_id(id).map(|p| (id, p))), - ) + pub fn seeds(&self, id: &Id) -> Vec<(NodeId, &Session)> { + if let Ok(seeds) = self.routing.get(id) { + seeds + .into_iter() + .filter_map(|id| self.sessions.by_id(&id).map(|p| (id, p))) + .collect() } else { - Box::new(std::iter::empty()) + vec![] } } @@ -313,14 +313,13 @@ where &mut self.reactor } - pub fn lookup(&self, id: Id) -> Lookup { - Lookup { + pub fn lookup(&self, id: Id) -> Result { + let remote = self.routing.get(&id)?.iter().cloned().collect(); + + Ok(Lookup { local: self.storage.get(&self.node_id(), id).unwrap(), - remote: self - .routing - .get(&id) - .map_or(vec![], |r| r.iter().cloned().collect()), - } + remote, + }) } pub fn initialize(&mut self, time: LocalTime) { @@ -387,7 +386,7 @@ where return; } - let seeds = self.seeds(&id).collect::>(); + let seeds = self.seeds(&id); let seeds = if let Some(seeds) = NonEmpty::from_vec(seeds) { seeds } else { @@ -618,7 +617,8 @@ where } else { return Ok(false); } - self.process_inventory(&message.inventory, *node, git); + self.process_inventory(&message.inventory, *node, git) + .unwrap(); if relay { return Ok(true); @@ -747,18 +747,19 @@ where } /// Process a peer inventory announcement by updating our routing table. - fn process_inventory(&mut self, inventory: &Inventory, from: NodeId, remote: &Url) { + fn process_inventory( + &mut self, + inventory: &Inventory, + from: NodeId, + remote: &Url, + ) -> Result<(), routing::Error> { for proj_id in inventory { - let inventory = self - .routing - .entry(*proj_id) - .or_insert_with(|| HashSet::with_hasher(self.rng.clone().into())); - // TODO: Fire an event on routing update. - if inventory.insert(from) && self.config.is_tracking(proj_id) { + if self.routing.insert(*proj_id, from)? && self.config.is_tracking(proj_id) { self.storage.fetch(*proj_id, remote).unwrap(); } } + Ok(()) } //////////////////////////////////////////////////////////////////////////// @@ -808,11 +809,12 @@ pub trait ServiceState { /// Get service configuration. fn config(&self) -> &Config; /// Get reference to routing table. - fn routing(&self) -> &Routing; + fn routing(&self) -> &dyn routing::Store; } -impl ServiceState for Service +impl ServiceState for Service where + R: routing::Store, G: Signer, S: ReadStorage, { @@ -836,7 +838,7 @@ where &self.config } - fn routing(&self) -> &Routing { + fn routing(&self) -> &dyn routing::Store { &self.routing } } @@ -871,7 +873,7 @@ impl fmt::Display for DisconnectReason { } } -impl Iterator for Service { +impl Iterator for Service { type Item = reactor::Io; fn next(&mut self) -> Option { diff --git a/radicle-node/src/service/routing.rs b/radicle-node/src/service/routing.rs new file mode 100644 index 00000000..7a51fbcc --- /dev/null +++ b/radicle-node/src/service/routing.rs @@ -0,0 +1,196 @@ +use std::collections::HashSet; +use std::path::Path; + +use rusqlite as sql; +use thiserror::Error; + +use crate::prelude::{Id, NodeId}; + +/// An error occuring in peer-to-peer networking code. +#[derive(Error, Debug)] +pub enum Error { + /// An Internal error. + #[error("internal error: {0}")] + Internal(#[from] sql::Error), +} + +/// Persistent file storage for a routing table. +#[derive(Debug)] +pub struct Table { + db: sql::Connection, +} + +impl Table { + const SCHEMA: &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 db = sql::Connection::open(path)?; + db.execute(Self::SCHEMA, [])?; + + Ok(Self { db }) + } + + /// Create a new in-memory routing table. + pub fn memory() -> Result { + let db = sql::Connection::open_in_memory()?; + db.execute(Self::SCHEMA, [])?; + + Ok(Self { db }) + } +} + +/// Backing store for a routing table. +pub trait Store { + /// Get the nodes seeding the given id. + fn get(&self, id: &Id) -> Result, Error>; + /// Add a new node seeding the given id. + fn insert(&mut self, id: Id, node: NodeId) -> 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>; +} + +impl Store for Table { + fn get(&self, id: &Id) -> Result, Error> { + let mut stmt = self + .db + .prepare("SELECT (node) FROM routing WHERE resource = ?")?; + let mut rows = stmt.query([id])?; + let mut nodes = HashSet::new(); + + while let Ok(Some(row)) = rows.next() { + let field = row.get(0)?; + nodes.insert(field); + } + Ok(nodes) + } + + fn insert(&mut self, id: Id, node: NodeId) -> Result { + let updated = self.db.execute( + "INSERT INTO routing (resource, node, time) VALUES (?, ?, ?) ON CONFLICT DO NOTHING", + (id, node, 0), + )?; + + Ok(updated > 0) + } + + fn entries(&self) -> Result>, Error> { + let mut stmt = self + .db + .prepare("SELECT resource, node FROM routing ORDER BY resource")?; + let mut rows = stmt.query([])?; + let mut entries = Vec::new(); + + while let Ok(Some(row)) = rows.next() { + let id = row.get(0)?; + let node = row.get(1)?; + + entries.push((id, node)); + } + Ok(Box::new(entries.into_iter())) + } + + fn remove(&mut self, id: &Id, node: &NodeId) -> Result { + let deleted = self.db.execute( + "DELETE FROM routing WHERE resource = ? AND node = ?", + (id, node), + )?; + + Ok(deleted > 0) + } +} + +#[cfg(test)] +mod test { + use super::*; + use crate::test::arbitrary; + + #[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(); + + for id in &ids { + for node in &nodes { + assert!(db.insert(*id, *node).unwrap()); + } + } + + for id in &ids { + let seeds = db.get(id).unwrap(); + for node in &nodes { + assert!(seeds.contains(node)); + } + } + } + + #[test] + fn test_iter() { + let ids = arbitrary::set::(6..9); + let nodes = arbitrary::set::(6..9); + let mut db = Table::open(":memory:").unwrap(); + + for id in &ids { + for node in &nodes { + assert!(db.insert(*id, *node).unwrap()); + } + } + + let results = db.entries().unwrap().collect::>(); + assert_eq!(results.len(), ids.len() * nodes.len()); + + let mut results_ids = results.iter().map(|(id, _)| *id).collect::>(); + results_ids.dedup(); + + assert_eq!(results_ids.len(), ids.len(), "Entries are grouped by id"); + } + + #[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(); + + for id in &ids { + for node in &nodes { + db.insert(*id, *node).unwrap(); + } + } + + for id in &ids { + for node in &nodes { + assert!(db.remove(id, node).unwrap()); + } + } + + for id in &ids { + assert!(db.get(id).unwrap().is_empty()); + } + } + + #[test] + fn test_insert_duplicate() { + let id = arbitrary::gen::(1); + let node = arbitrary::gen::(1); + let mut db = Table::open(":memory:").unwrap(); + + assert!(db.insert(id, node).unwrap()); + assert!(!db.insert(id, node).unwrap()); + assert!(!db.insert(id, node).unwrap()); + } + + #[test] + fn test_remove_redundant() { + let id = arbitrary::gen::(1); + let node = arbitrary::gen::(1); + let mut db = Table::open(":memory:").unwrap(); + + assert!(db.insert(id, node).unwrap()); + assert!(db.remove(&id, &node).unwrap()); + assert!(!db.remove(&id, &node).unwrap()); + } +} diff --git a/radicle-node/src/service/routing/schema.sql b/radicle-node/src/service/routing/schema.sql new file mode 100644 index 00000000..bf0486c8 --- /dev/null +++ b/radicle-node/src/service/routing/schema.sql @@ -0,0 +1,15 @@ +-- +-- Routing table SQL schema. +-- +create table if not exists "routing" ( + "id" integer primary key, + -- 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, + + unique("resource", "node") +); +create index "routing_index" on routing ("resource", "node"); diff --git a/radicle-node/src/test/handle.rs b/radicle-node/src/test/handle.rs index a02dd9f3..deb33e3a 100644 --- a/radicle-node/src/test/handle.rs +++ b/radicle-node/src/test/handle.rs @@ -36,7 +36,7 @@ impl traits::Handle for Handle { Ok(()) } - fn routing(&self) -> Result)>, Error> { + fn routing(&self) -> Result, Error> { unimplemented!(); } diff --git a/radicle-node/src/test/peer.rs b/radicle-node/src/test/peer.rs index cc14fce5..0da3ede5 100644 --- a/radicle-node/src/test/peer.rs +++ b/radicle-node/src/test/peer.rs @@ -9,6 +9,7 @@ use crate::clock::{RefClock, Timestamp}; use crate::collections::HashMap; use crate::git; use crate::git::Url; +use crate::prelude::NodeId; use crate::service; use crate::service::config::*; use crate::service::message::*; @@ -20,7 +21,8 @@ use crate::test::simulator; use crate::{Link, LocalTime}; /// Service instantiation used for testing. -pub type Service = service::Service, S, MockSigner>; +pub type Service = + service::Service, S, MockSigner>; #[derive(Debug)] pub struct Peer { @@ -101,7 +103,8 @@ where let local_time = LocalTime::now(); let clock = RefClock::from(local_time); let signer = MockSigner::new(&mut rng); - let service = Service::new(config, clock, storage, addrs, signer, rng.clone()); + let routing = routing::Table::memory().unwrap(); + let service = Service::new(config, clock, routing, storage, addrs, signer, rng.clone()); let ip = ip.into(); let local_addr = net::SocketAddr::new(ip, rng.u16(..)); diff --git a/radicle-node/src/test/tests.rs b/radicle-node/src/test/tests.rs index e3e5291c..49840fee 100644 --- a/radicle-node/src/test/tests.rs +++ b/radicle-node/src/test/tests.rs @@ -527,7 +527,7 @@ fn prop_inventory_exchange_dense() { let alice = Peer::new("alice", [7, 7, 7, 7], alice_inv.clone()); let mut bob = Peer::new("bob", [8, 8, 8, 8], bob_inv.clone()); let mut eve = Peer::new("eve", [9, 9, 9, 9], eve_inv.clone()); - let mut routing = Routing::with_hasher(rng.clone().into()); + let mut routing = HashMap::with_hasher(rng.clone().into()); for (inv, peer) in &[ (alice_inv.inventory, alice.node_id()), @@ -562,7 +562,7 @@ fn prop_inventory_exchange_dense() { for (proj_id, remotes) in &routing { for peer in peers.values() { - let lookup = peer.lookup(*proj_id); + let lookup = peer.lookup(*proj_id).unwrap(); if lookup.local.is_some() { peer.get(*proj_id) diff --git a/radicle-node/src/transport.rs b/radicle-node/src/transport.rs index 12934162..473ef4ef 100644 --- a/radicle-node/src/transport.rs +++ b/radicle-node/src/transport.rs @@ -8,6 +8,7 @@ use nakamoto_net::{Io, Link}; use crate::address_book; use crate::collections::HashMap; use crate::crypto; +use crate::service::routing; use crate::service::{Command, DisconnectReason, Event, Service}; use crate::storage::WriteStorage; use crate::wire::Wire; @@ -18,13 +19,13 @@ struct Peer { } #[derive(Debug)] -pub struct Transport { +pub struct Transport { peers: HashMap, - inner: Wire, + inner: Wire, } -impl Transport { - pub fn new(inner: Wire) -> Self { +impl Transport { + pub fn new(inner: Wire) -> Self { Self { peers: HashMap::default(), inner, @@ -32,8 +33,9 @@ impl Transport { } } -impl nakamoto::Protocol for Transport +impl nakamoto::Protocol for Transport where + R: routing::Store, T: WriteStorage + 'static, S: address_book::Store, G: crypto::Signer, @@ -84,7 +86,7 @@ where } } -impl Iterator for Transport { +impl Iterator for Transport { type Item = Io; fn next(&mut self) -> Option { @@ -92,15 +94,15 @@ impl Iterator for Transport { } } -impl Deref for Transport { - type Target = Service; +impl Deref for Transport { + type Target = Service; fn deref(&self) -> &Self::Target { &self.inner } } -impl DerefMut for Transport { +impl DerefMut for Transport { fn deref_mut(&mut self) -> &mut Self::Target { &mut self.inner } diff --git a/radicle-node/src/wire.rs b/radicle-node/src/wire.rs index f3886d1e..fb7a5ff1 100644 --- a/radicle-node/src/wire.rs +++ b/radicle-node/src/wire.rs @@ -19,8 +19,8 @@ use crate::git::fmt; use crate::hash::Digest; use crate::identity::Id; use crate::service; -use crate::service::filter; use crate::service::reactor::Io; +use crate::service::{filter, routing}; use crate::storage::refs::Refs; use crate::storage::refs::SignedRefs; use crate::storage::WriteStorage; @@ -437,13 +437,13 @@ impl Decode for SignedRefs { } #[derive(Debug)] -pub struct Wire { +pub struct Wire { inboxes: HashMap, - inner: service::Service, + inner: service::Service, } -impl Wire { - pub fn new(inner: service::Service) -> Self { +impl Wire { + pub fn new(inner: service::Service) -> Self { Self { inboxes: HashMap::new(), inner, @@ -451,8 +451,9 @@ impl Wire { } } -impl Wire +impl Wire where + R: routing::Store, S: address_book::Store, T: WriteStorage + 'static, G: Signer, @@ -501,7 +502,7 @@ where } } -impl Iterator for Wire { +impl Iterator for Wire { type Item = nakamoto::Io; fn next(&mut self) -> Option { @@ -526,15 +527,15 @@ impl Iterator for Wire { } } -impl Deref for Wire { - type Target = service::Service; +impl Deref for Wire { + type Target = service::Service; fn deref(&self) -> &Self::Target { &self.inner } } -impl DerefMut for Wire { +impl DerefMut for Wire { fn deref_mut(&mut self) -> &mut Self::Target { &mut self.inner } diff --git a/radicle/Cargo.toml b/radicle/Cargo.toml index f9d72260..3248df17 100644 --- a/radicle/Cargo.toml +++ b/radicle/Cargo.toml @@ -8,6 +8,7 @@ edition = "2021" [features] default = [] test = ["quickcheck"] +sql = ["rusqlite"] [dependencies] base64 = { version= "0.13" } @@ -27,6 +28,7 @@ serde = { version = "1", features = ["derive"] } serde_json = { version = "1", features = ["preserve_order"] } siphasher = { version = "0.3.10" } radicle-git-ext = { version = "0", features = ["serde"] } +rusqlite = { version = "0.28.0", features = ["bundled"], optional = true } nonempty = { version = "0.8.0", features = ["serialize"] } tempfile = { version = "3.3.0" } thiserror = { version = "1" } diff --git a/radicle/src/lib.rs b/radicle/src/lib.rs index 0da24f19..43ac14d6 100644 --- a/radicle/src/lib.rs +++ b/radicle/src/lib.rs @@ -8,6 +8,8 @@ pub mod node; pub mod profile; pub mod rad; pub mod serde_ext; +#[cfg(feature = "sql")] +pub mod sql; pub mod ssh; pub mod storage; #[cfg(feature = "test")] diff --git a/radicle/src/sql.rs b/radicle/src/sql.rs new file mode 100644 index 00000000..74da2287 --- /dev/null +++ b/radicle/src/sql.rs @@ -0,0 +1,48 @@ +use std::str; +use std::str::FromStr; + +use rusqlite as sql; +use rusqlite::types::{FromSql, FromSqlError, FromSqlResult, ToSql, ToSqlOutput, ValueRef}; + +use crate::crypto::PublicKey; +use crate::identity::Id; + +impl FromSql for Id { + fn column_result(value: ValueRef<'_>) -> FromSqlResult { + match value { + ValueRef::Text(id) => { + let id = str::from_utf8(id).map_err(|e| FromSqlError::Other(Box::new(e)))?; + let id = Id::from_str(id).map_err(|e| FromSqlError::Other(Box::new(e)))?; + + Ok(id) + } + _ => Err(FromSqlError::InvalidType), + } + } +} + +impl ToSql for Id { + fn to_sql(&self) -> sql::Result> { + Ok(ToSqlOutput::from(self.to_string())) + } +} + +impl FromSql for PublicKey { + fn column_result(value: ValueRef<'_>) -> FromSqlResult { + match value { + ValueRef::Text(pk) => { + let pk = str::from_utf8(pk).map_err(|e| FromSqlError::Other(Box::new(e)))?; + let pk = PublicKey::from_str(pk).map_err(|e| FromSqlError::Other(Box::new(e)))?; + + Ok(pk) + } + _ => Err(FromSqlError::InvalidType), + } + } +} + +impl ToSql for PublicKey { + fn to_sql(&self) -> sql::Result> { + Ok(ToSqlOutput::from(self.to_string())) + } +}