From 947f1bfc018a641aded13e7c0b4ab526cbd9b117 Mon Sep 17 00:00:00 2001 From: cloudhead Date: Wed, 8 Nov 2023 15:08:55 +0100 Subject: [PATCH] node: Use variable subscribe backlog delta Previously, we would always ask nodes for messages from one hour ago. This meant that we could miss messages if our node was offline for more than an hour. This change uses a variable backlog by checking the timestamp of the last gossip message we received. --- radicle-node/src/service.rs | 42 +++++++++++++++++++----- radicle-node/src/service/gossip.rs | 29 ---------------- radicle-node/src/service/gossip/store.rs | 14 ++++++++ 3 files changed, 47 insertions(+), 38 deletions(-) diff --git a/radicle-node/src/service.rs b/radicle-node/src/service.rs index e80a105f..254678a3 100644 --- a/radicle-node/src/service.rs +++ b/radicle-node/src/service.rs @@ -73,8 +73,9 @@ pub const KEEP_ALIVE_DELTA: LocalDuration = LocalDuration::from_mins(1); pub const MAX_TIME_DELTA: LocalDuration = LocalDuration::from_mins(60); /// Maximum attempts to connect to a peer before we give up. pub const MAX_CONNECTION_ATTEMPTS: usize = 3; -/// How far back from the present time should we request gossip messages when connecting to a peer. -pub const SUBSCRIBE_BACKLOG_DELTA: LocalDuration = LocalDuration::from_mins(60); +/// How far back from the present time should we request gossip messages when connecting to a peer, +/// when we initially come online. +pub const INITIAL_SUBSCRIBE_BACKLOG_DELTA: LocalDuration = LocalDuration::from_mins(60 * 24); /// Minimum amount of time to wait before reconnecting to a peer. pub const MIN_RECONNECTION_DELTA: LocalDuration = LocalDuration::from_secs(3); /// Maximum amount of time to wait before reconnecting to a peer. @@ -1232,17 +1233,40 @@ where /// Set of initial messages to send to a peer. fn initial(&self, _link: Link) -> Vec { let filter = self.filter(); + let now = self.clock(); + let inventory = match self.storage.inventory() { + Ok(i) => i, + Err(e) => { + // Other than crashing the node completely, there's nothing we can do + // here besides returning an empty inventory and logging an error. + error!(target: "service", "Error getting local inventory for initial messages: {e}"); + vec![] + } + }; // TODO: Only subscribe to outbound connections, otherwise we will consume too // much bandwidth. - gossip::handshake( - self.node.clone(), - self.clock.as_millis(), - &self.storage, - &self.signer, - filter, - ) + // If we've been previously connected to the network, we'll have received gossip messages. + // Instead of simply taking the last timestamp we try to ensure we don't miss any + // messages due un-synchronized clocks. + // + // 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() { + Ok(Some(last)) => last - MAX_TIME_DELTA.as_millis() as Timestamp, + Ok(None) => (*now - INITIAL_SUBSCRIBE_BACKLOG_DELTA).as_millis() as Timestamp, + Err(e) => { + error!(target: "service", "Error getting the lastest gossip message from storage: {e}"); + return vec![]; + } + }; + + vec![ + Message::node(self.node.clone(), &self.signer), + Message::inventory(gossip::inventory(now.as_millis(), inventory), &self.signer), + Message::subscribe(filter, since, Timestamp::MAX), + ] } /// Update our routing table with our local node's inventory. diff --git a/radicle-node/src/service/gossip.rs b/radicle-node/src/service/gossip.rs index 3a14cd4e..02cd159f 100644 --- a/radicle-node/src/service/gossip.rs +++ b/radicle-node/src/service/gossip.rs @@ -1,39 +1,10 @@ pub mod store; use super::*; -use crate::service::filter::Filter; pub use store::Error; pub use store::GossipStore as Store; -pub fn handshake( - node: NodeAnnouncement, - now: Timestamp, - storage: &S, - signer: &G, - filter: Filter, -) -> Vec { - let inventory = match storage.inventory() { - Ok(i) => i, - Err(e) => { - error!("Error getting local inventory for handshake: {}", e); - // Other than crashing the node completely, there's nothing we can do - // here besides returning an empty inventory and logging an error. - vec![] - } - }; - - vec![ - Message::node(node, signer), - Message::inventory(gossip::inventory(now, inventory), signer), - Message::subscribe( - filter, - now - SUBSCRIBE_BACKLOG_DELTA.as_millis() as u64, - Timestamp::MAX, - ), - ] -} - pub fn node(config: &Config, timestamp: Timestamp) -> NodeAnnouncement { let features = config.features(); let alias = config.alias.clone(); diff --git a/radicle-node/src/service/gossip/store.rs b/radicle-node/src/service/gossip/store.rs index 5fbc41af..2e4057b2 100644 --- a/radicle-node/src/service/gossip/store.rs +++ b/radicle-node/src/service/gossip/store.rs @@ -57,6 +57,20 @@ impl GossipStore { Ok(self.db.change_count()) } + /// Get the last announcement in the store, by timestamp. + pub fn last(&self) -> Result, Error> { + let stmt = self + .db + .prepare("SELECT MAX(timestamp) AS latest FROM `announcements`")?; + + if let Some(Ok(row)) = stmt.into_iter().next() { + let latest = row.try_read::, _>(0)?; + + return Ok(latest.map(|l| l as Timestamp)); + } + 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 {