From 9a82aae993c8f927222833d5a8b47983af293720 Mon Sep 17 00:00:00 2001
From: cloudhead <cloudhead@radicle.xyz>
Date: Sun, 5 May 2024 22:35:56 +0200
Subject: [PATCH] node: Reset connection attempts when stable

Instead of resetting the number of attempts on connection, reset it when
the connection is stable, ie. after some time has passed.

This prevents connection retries from happening in succession when a
peer is disconnected during the initial handshake, as is the case with
invalid subscribe timestamps.
---
 radicle-node/src/service.rs         | 28 +++++++++++++++++++--------
 radicle-node/src/service/session.rs | 30 ++++++++++++++++++++++++++---
 radicle/src/node.rs                 |  4 ++++
 3 files changed, 51 insertions(+), 11 deletions(-)

diff --git a/radicle-node/src/service.rs b/radicle-node/src/service.rs
index 0c4b5855..5e8697b1 100644
--- a/radicle-node/src/service.rs
+++ b/radicle-node/src/service.rs
@@ -719,6 +719,7 @@ where
 
             self.keep_alive(&now);
             self.disconnect_unresponsive_peers(&now);
+            self.idle_connections();
             self.maintain_connections();
             self.outbox.wakeup(IDLE_INTERVAL);
             self.last_idle = now;
@@ -1202,14 +1203,6 @@ where
             if let Some(peer) = self.sessions.get_mut(&remote) {
                 peer.to_connected(self.clock);
                 self.outbox.write_all(peer, msgs);
-
-                if let Err(e) =
-                    self.db
-                        .addresses_mut()
-                        .connected(&remote, &peer.addr, self.clock.into())
-                {
-                    error!(target: "service", "Error updating address book with connection: {e}");
-                }
             }
         } else {
             match self.sessions.entry(remote) {
@@ -2255,6 +2248,25 @@ where
         Ok(())
     }
 
+    /// Run idle task for all connections.
+    fn idle_connections(&mut self) {
+        for (_, sess) in self.sessions.iter_mut() {
+            sess.idle(self.clock);
+
+            if sess.is_stable() {
+                // Mark as connected once connection is stable.
+                if let Err(e) =
+                    self.db
+                        .addresses_mut()
+                        .connected(&sess.id, &sess.addr, self.clock.into())
+                {
+                    error!(target: "service", "Error updating address book with connection: {e}");
+                }
+            }
+        }
+    }
+
+    /// Try to maintain a target number of connections.
     fn maintain_connections(&mut self) {
         let PeerConfig::Dynamic { target } = self.config.peers else {
             return;
diff --git a/radicle-node/src/service/session.rs b/radicle-node/src/service/session.rs
index f5638b54..fdf75138 100644
--- a/radicle-node/src/service/session.rs
+++ b/radicle-node/src/service/session.rs
@@ -5,11 +5,14 @@ use crate::node::config::Limits;
 use crate::node::Severity;
 use crate::service::message;
 use crate::service::message::Message;
-use crate::service::{Address, LocalTime, NodeId, Outbox, RepoId, Rng};
+use crate::service::{Address, LocalDuration, LocalTime, NodeId, Outbox, RepoId, Rng};
 use crate::{Link, Timestamp};
 
 pub use crate::node::{PingState, State};
 
+/// Time after which a connection is considered stable.
+pub const CONNECTION_STABLE_THRESHOLD: LocalDuration = LocalDuration::from_mins(1);
+
 #[derive(thiserror::Error, Debug, Clone, Copy)]
 pub enum Error {
     /// The remote peer sent an invalid announcement timestamp,
@@ -61,7 +64,7 @@ pub struct Session {
 
     /// Connection attempts. For persistent peers, Tracks
     /// how many times we've attempted to connect. We reset this to zero
-    /// upon successful connection.
+    /// upon successful connection, once the connection is stable.
     attempts: usize,
     /// Source of entropy.
     rng: Rng,
@@ -120,6 +123,7 @@ impl Session {
                 ping: PingState::default(),
                 fetching: HashSet::default(),
                 latencies: VecDeque::default(),
+                stable: false,
             },
             link: Link::Inbound,
             subscribe: None,
@@ -135,6 +139,10 @@ impl Session {
         matches!(self.state, State::Attempted { .. })
     }
 
+    pub fn is_stable(&self) -> bool {
+        matches!(self.state, State::Connected { stable: true, .. })
+    }
+
     pub fn is_connected(&self) -> bool {
         self.state.is_connected()
     }
@@ -167,6 +175,22 @@ impl Session {
         self.attempts
     }
 
+    /// Run 'idle' task for session.
+    pub fn idle(&mut self, now: LocalTime) {
+        if let State::Connected {
+            since,
+            ref mut stable,
+            ..
+        } = self.state
+        {
+            if now >= since && now.duration_since(since) >= CONNECTION_STABLE_THRESHOLD {
+                *stable = true;
+                // Reset number of attempts for stable connections.
+                self.attempts = 0;
+            }
+        }
+    }
+
     /// Mark this session as fetching the given RID.
     ///
     /// # Panics
@@ -204,7 +228,6 @@ impl Session {
     }
 
     pub fn to_connected(&mut self, since: LocalTime) {
-        self.attempts = 0;
         self.last_active = since;
 
         let State::Attempted = &self.state else {
@@ -215,6 +238,7 @@ impl Session {
             ping: PingState::default(),
             fetching: HashSet::default(),
             latencies: VecDeque::default(),
+            stable: false,
         };
     }
 
diff --git a/radicle/src/node.rs b/radicle/src/node.rs
index c62f663c..d19cd586 100644
--- a/radicle/src/node.rs
+++ b/radicle/src/node.rs
@@ -105,6 +105,9 @@ pub enum State {
         /// Measured latencies for this peer.
         #[serde(skip)]
         latencies: VecDeque<LocalDuration>,
+        /// Whether the connection is stable.
+        #[serde(skip)]
+        stable: bool,
     },
     /// When a peer is disconnected.
     #[serde(rename_all = "camelCase")]
@@ -1260,6 +1263,7 @@ mod test {
                 ping: Default::default(),
                 fetching: Default::default(),
                 latencies: VecDeque::default(),
+                stable: false,
             }))
             .unwrap(),
         )