From 267aba59016e312039a0724a1e98f89712b2adfc Mon Sep 17 00:00:00 2001 From: Alexis Sellier Date: Wed, 12 Apr 2023 15:38:43 +0200 Subject: [PATCH] node: Announce inventory only if it changed Previously we were announcing inventories all the time, due to relying on the timestamp changing on insertion. We change this to only announce our inventory when it actually changes. --- radicle-node/src/address/store.rs | 2 +- radicle-node/src/lib.rs | 1 - radicle-node/src/service.rs | 83 ++++++++++++++++++++----------- radicle-node/src/sql.rs | 21 -------- radicle/src/node/routing.rs | 80 +++++++++++++++++++++-------- radicle/src/sql.rs | 20 ++++++++ 6 files changed, 133 insertions(+), 74 deletions(-) delete mode 100644 radicle-node/src/sql.rs diff --git a/radicle-node/src/address/store.rs b/radicle-node/src/address/store.rs index 9e48a7f6..ce8fc14e 100644 --- a/radicle-node/src/address/store.rs +++ b/radicle-node/src/address/store.rs @@ -4,13 +4,13 @@ use std::{fmt, io}; use radicle::node; use radicle::node::Address; use radicle::prelude::Timestamp; +use radicle::sql::transaction; use sqlite as sql; use thiserror::Error; use crate::address::types; use crate::address::{KnownAddress, Source}; use crate::service::NodeId; -use crate::sql::transaction; use crate::wire::AddressType; #[derive(Error, Debug)] diff --git a/radicle-node/src/lib.rs b/radicle-node/src/lib.rs index d434d2f4..12b4df9a 100644 --- a/radicle-node/src/lib.rs +++ b/radicle-node/src/lib.rs @@ -6,7 +6,6 @@ pub mod logger; pub mod runtime; pub mod service; pub mod signals; -pub mod sql; #[cfg(any(test, feature = "test"))] pub mod test; #[cfg(test)] diff --git a/radicle-node/src/service.rs b/radicle-node/src/service.rs index 81080c48..5deaab06 100644 --- a/radicle-node/src/service.rs +++ b/radicle-node/src/service.rs @@ -27,6 +27,7 @@ use crate::identity::IdentityError; use crate::identity::{Doc, Id}; use crate::node; use crate::node::routing; +use crate::node::routing::InsertResult; use crate::node::{Address, Features, FetchResult, Seed, Seeds}; use crate::prelude::*; use crate::runtime::Emitter; @@ -102,6 +103,23 @@ pub enum Event { }, } +/// Result of syncing our routing table with a node's inventory. +#[derive(Default)] +struct SyncedRouting { + /// Repo entries added. + added: Vec, + /// Repo entries removed. + removed: Vec, + /// Repo entries updated (time). + updated: Vec, +} + +impl SyncedRouting { + fn is_empty(&self) -> bool { + self.added.is_empty() && self.removed.is_empty() && self.updated.is_empty() + } +} + /// General service error. #[derive(thiserror::Error, Debug)] pub enum Error { @@ -527,10 +545,11 @@ where } } Command::SyncInventory(resp) => { - let updated = self + let synced = self .sync_inventory() .expect("Service::command: error syncing inventory"); - resp.send(!updated.is_empty()).ok(); + resp.send(synced.added.len() + synced.removed.len() > 0) + .ok(); } Command::QueryState(query, sender) => { sender.send(query(self)).ok(); @@ -806,8 +825,8 @@ where } match self.sync_routing(&message.inventory, *announcer, message.timestamp) { - Ok(updated) => { - if updated.is_empty() { + Ok(synced) => { + if synced.is_empty() { debug!(target: "service", "No routes updated by inventory announcement from {announcer}"); return Ok(false); } @@ -865,11 +884,11 @@ where // We update inventories when receiving ref announcements, as these could come // from a new repository being initialized. - if let Ok(updated) = self + if let Ok(result) = self .routing .insert(message.rid, *announcer, message.timestamp) { - if updated { + if let InsertResult::SeedAdded = result { self.emitter.emit(Event::SeedDiscovered { rid: message.rid, nid: *relayer, @@ -1113,11 +1132,11 @@ where } /// Update our routing table with our local node's inventory. - fn sync_inventory(&mut self) -> Result, Error> { + fn sync_inventory(&mut self) -> Result { let inventory = self.storage.inventory()?; - let updated = self.sync_routing(&inventory, self.node_id(), self.time())?; + let result = self.sync_routing(&inventory, self.node_id(), self.time())?; - Ok(updated) + Ok(result) } /// Process a peer inventory announcement by updating our routing table. @@ -1128,39 +1147,43 @@ where inventory: &[Id], from: NodeId, timestamp: Timestamp, - ) -> Result, Error> { - let mut updated = Vec::new(); + ) -> Result { + let mut synced = SyncedRouting::default(); let mut included = HashSet::new(); for rid in inventory { included.insert(rid); - if self.routing.insert(*rid, from, timestamp)? { - info!(target: "service", "Routing table updated for {rid} with seed {from}"); - self.emitter.emit(Event::SeedDiscovered { - rid: *rid, - nid: from, - }); + match self.routing.insert(*rid, from, timestamp)? { + InsertResult::SeedAdded => { + info!(target: "service", "Routing table updated for {rid} with seed {from}"); + self.emitter.emit(Event::SeedDiscovered { + rid: *rid, + nid: from, + }); - if self - .tracking - .is_repo_tracked(rid) - .expect("Service::process_inventory: error accessing tracking configuration") - { - // TODO: We should fetch here if we're already connected, case this seed has - // refs we don't have. + if self.tracking.is_repo_tracked(rid).expect( + "Service::process_inventory: error accessing tracking configuration", + ) { + // TODO: We should fetch here if we're already connected, case this seed has + // refs we don't have. + } + synced.added.push(*rid); } - updated.push(*rid); + InsertResult::TimeUpdated => { + synced.updated.push(*rid); + } + InsertResult::NotUpdated => {} } } for rid in self.routing.get_resources(&from)?.into_iter() { if !included.contains(&rid) { if self.routing.remove(&rid, &from)? { - updated.push(rid); + synced.removed.push(rid); self.emitter.emit(Event::SeedDropped { rid, nid: from }); } } } - Ok(updated) + Ok(synced) } /// Announce local refs for given id. @@ -1214,9 +1237,9 @@ where fn sync_and_announce(&mut self) { match self.sync_inventory() { - // TODO: This will always be `true` because the local clock is ticking. - Ok(updated) => { - if !updated.is_empty() { + Ok(synced) => { + // Only announce if our inventory changed. + if synced.added.len() + synced.removed.len() > 0 { if let Err(e) = self .storage .inventory() diff --git a/radicle-node/src/sql.rs b/radicle-node/src/sql.rs deleted file mode 100644 index 81d80850..00000000 --- a/radicle-node/src/sql.rs +++ /dev/null @@ -1,21 +0,0 @@ -use sqlite as sql; - -/// Run an SQL query inside a transaction. -/// Commits the transaction on success, and rolls back on error. -pub fn transaction( - db: &sql::Connection, - query: impl FnOnce(&sql::Connection) -> Result, -) -> Result { - db.execute("BEGIN")?; - - match query(db) { - Ok(result) => { - db.execute("COMMIT")?; - Ok(result) - } - Err(err) => { - db.execute("ROLLBACK")?; - Err(err) - } - } -} diff --git a/radicle/src/node/routing.rs b/radicle/src/node/routing.rs index 09fe5bc1..3ab3363e 100644 --- a/radicle/src/node/routing.rs +++ b/radicle/src/node/routing.rs @@ -8,6 +8,7 @@ use thiserror::Error; use crate::{ prelude::Timestamp, prelude::{Id, NodeId}, + sql::transaction, }; /// How long to wait for the database lock to be released before failing a read. @@ -15,6 +16,17 @@ 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 { + /// Nothing was updated. + NotUpdated, + /// The entry's timestamp was updated. + TimeUpdated, + /// A new entry was inserted. + SeedAdded, +} + /// An error occuring in peer-to-peer networking code. #[derive(Error, Debug)] pub enum Error { @@ -83,7 +95,7 @@ pub trait Store { Ok(self.len()? == 0) } /// Add a new node seeding the given id. - fn insert(&mut self, id: Id, node: NodeId, time: Timestamp) -> Result; + fn insert(&mut self, id: Id, node: NodeId, time: Timestamp) -> Result; /// Remove a node for the given id. fn remove(&mut self, id: &Id, node: &NodeId) -> Result; /// Iterate over all entries in the routing table. @@ -135,22 +147,37 @@ impl Store for Table { Ok(None) } - fn insert(&mut self, id: Id, node: NodeId, time: Timestamp) -> Result { + fn insert(&mut self, id: Id, node: NodeId, time: Timestamp) -> Result { let time: i64 = time.try_into().map_err(|_| Error::UnitOverflow)?; - let mut stmt = self.db.prepare( - "INSERT INTO routing (resource, node, time) - VALUES (?, ?, ?) - ON CONFLICT DO UPDATE - SET time = ?3 - WHERE time < ?3", - )?; - stmt.bind((1, &id))?; - stmt.bind((2, &node))?; - stmt.bind((3, time))?; - stmt.next()?; + transaction(&self.db, |db| { + let mut stmt = + db.prepare("SELECT (time) FROM routing WHERE resource = ? AND node = ?")?; - Ok(self.db.change_count() > 0) + 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) + VALUES (?, ?, ?) + ON CONFLICT DO UPDATE + SET time = ?3 + WHERE time < ?3", + )?; + + stmt.bind((1, &id))?; + stmt.bind((2, &node))?; + stmt.bind((3, time))?; + stmt.next()?; + + Ok(match (self.db.change_count() > 0, existed) { + (true, true) => InsertResult::TimeUpdated, + (true, false) => InsertResult::SeedAdded, + (false, _) => InsertResult::NotUpdated, + }) + }) + .map_err(Error::from) } fn entries(&self) -> Result>, Error> { @@ -226,7 +253,7 @@ mod test { for id in &ids { for node in &nodes { - assert!(db.insert(*id, *node, 0).unwrap()); + assert_eq!(db.insert(*id, *node, 0).unwrap(), InsertResult::SeedAdded); } } @@ -246,7 +273,7 @@ mod test { for id in &ids { for node in &nodes { - assert!(db.insert(*id, *node, 0).unwrap()); + assert_eq!(db.insert(*id, *node, 0).unwrap(), InsertResult::SeedAdded); } } @@ -266,7 +293,7 @@ mod test { for id in &ids { for node in &nodes { - assert!(db.insert(*id, *node, 0).unwrap()); + assert_eq!(db.insert(*id, *node, 0).unwrap(), InsertResult::SeedAdded); } } @@ -306,9 +333,20 @@ mod test { let node = arbitrary::gen::(1); let mut db = Table::open(":memory:").unwrap(); - assert!(db.insert(id, node, 0).unwrap()); - assert!(!db.insert(id, node, 0).unwrap()); - assert!(!db.insert(id, node, 0).unwrap()); + assert_eq!(db.insert(id, node, 0).unwrap(), InsertResult::SeedAdded); + assert_eq!(db.insert(id, node, 0).unwrap(), InsertResult::NotUpdated); + assert_eq!(db.insert(id, node, 0).unwrap(), InsertResult::NotUpdated); + } + + #[test] + fn test_insert_existing_updated_time() { + let id = arbitrary::gen::(1); + let node = arbitrary::gen::(1); + let mut db = Table::open(":memory:").unwrap(); + + assert_eq!(db.insert(id, node, 0).unwrap(), InsertResult::SeedAdded); + assert_eq!(db.insert(id, node, 1).unwrap(), InsertResult::TimeUpdated); + assert_eq!(db.entry(&id, &node).unwrap(), Some(1)); } #[test] @@ -317,7 +355,7 @@ mod test { let node = arbitrary::gen::(1); let mut db = Table::open(":memory:").unwrap(); - assert!(db.insert(id, node, 0).unwrap()); + assert_eq!(db.insert(id, node, 0).unwrap(), InsertResult::SeedAdded); assert!(db.remove(&id, &node).unwrap()); assert!(!db.remove(&id, &node).unwrap()); } diff --git a/radicle/src/sql.rs b/radicle/src/sql.rs index 147c95bc..0885bac6 100644 --- a/radicle/src/sql.rs +++ b/radicle/src/sql.rs @@ -8,6 +8,26 @@ use crate::identity::Id; use crate::node; use crate::node::Address; +/// Run an SQL query inside a transaction. +/// Commits the transaction on success, and rolls back on error. +pub fn transaction( + db: &sql::Connection, + query: impl FnOnce(&sql::Connection) -> Result, +) -> Result { + db.execute("BEGIN")?; + + match query(db) { + Ok(result) => { + db.execute("COMMIT")?; + Ok(result) + } + Err(err) => { + db.execute("ROLLBACK")?; + Err(err) + } + } +} + impl TryFrom<&Value> for Id { type Error = sql::Error;