From a74cb416c1661f5f9899fe34a02f26b26c85000a Mon Sep 17 00:00:00 2001 From: Slack Coder Date: Thu, 3 Nov 2022 13:49:59 -0500 Subject: [PATCH] node: sync project routing for known nodes Keep the node's understanding of project hosting location current by updating how Inventory Announcement messages are processed. The Inventory Announcements contain a node's entire listing of projects. Replace the routing table information for the announcer with those from the message. Signed-off-by: Slack Coder --- radicle-node/src/service.rs | 12 +++++++--- radicle-node/src/service/routing.rs | 35 +++++++++++++++++++++++++++++ 2 files changed, 44 insertions(+), 3 deletions(-) diff --git a/radicle-node/src/service.rs b/radicle-node/src/service.rs index c344c830..2bec7a9a 100644 --- a/radicle-node/src/service.rs +++ b/radicle-node/src/service.rs @@ -6,7 +6,7 @@ pub mod routing; pub mod session; use std::collections::hash_map::Entry; -use std::collections::{BTreeMap, HashMap}; +use std::collections::{BTreeMap, HashMap, HashSet}; use std::ops::{Deref, DerefMut}; use std::sync::Arc; use std::{fmt, net, net::IpAddr, str}; @@ -876,17 +876,23 @@ where /// Process a peer inventory announcement by updating our routing table. fn process_inventory( &mut self, - inventory: &Inventory, + inventory: &Vec, from: NodeId, timestamp: &Timestamp, ) -> Result<(), Error> { + let mut included = HashSet::new(); for proj_id in inventory { - // TODO: Fire an event on routing update. + included.insert(proj_id); if self.routing.insert(*proj_id, from, *timestamp)? && self.config.is_tracking(proj_id) { log::info!("Routing table updated for {} with seed {}", proj_id, from); } } + for id in self.routing.get_resources(&from)?.into_iter() { + if !included.contains(&id) { + self.routing.remove(&id, &from)?; + } + } Ok(()) } diff --git a/radicle-node/src/service/routing.rs b/radicle-node/src/service/routing.rs index cebbb917..cb8ca128 100644 --- a/radicle-node/src/service/routing.rs +++ b/radicle-node/src/service/routing.rs @@ -57,6 +57,8 @@ impl Table { pub trait Store { /// Get the nodes seeding the given id. fn get(&self, id: &Id) -> Result, Error>; + /// Get the resources seeded by the given node. + fn get_resources(&self, node_id: &NodeId) -> Result, Error>; /// Get a specific entry. fn entry(&self, id: &Id, node: &NodeId) -> Result, Error>; /// Add a new node seeding the given id. @@ -83,6 +85,19 @@ impl Store for Table { Ok(nodes) } + fn get_resources(&self, node: &NodeId) -> Result, Error> { + let mut stmt = self + .db + .prepare("SELECT resource FROM routing WHERE node = ?")?; + stmt.bind(1, node)?; + + let mut resources = HashSet::new(); + for row in stmt.into_cursor() { + resources.insert(row?.get::("resource")); + } + Ok(resources) + } + fn entry(&self, id: &Id, node: &NodeId) -> Result, Error> { let mut stmt = self .db @@ -179,6 +194,26 @@ mod test { } } + #[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(); + + for id in &ids { + for node in &nodes { + assert!(db.insert(*id, *node, 0).unwrap()); + } + } + + for node in &nodes { + let projects = db.get_resources(node).unwrap(); + for id in &ids { + assert!(projects.contains(id)); + } + } + } + #[test] fn test_entries() { let ids = arbitrary::set::(6..9);