node: Fix attempted session state

We shouldn't be marking attempted peers as "connected". We should wait
for the connection to be properly established.

Signed-off-by: Alexis Sellier <alexis@radicle.xyz>

radicle-node/src/service.rs

@@ -520,10 +520,10 @@ where
         debug!("Attempted connection to {id} ({addr})");
 
         let persistent = self.config.is_persistent(&id);
-        let peer = self.sessions.entry(id).or_insert_with(|| {
-            Session::new(id, Link::Outbound, persistent, self.rng.clone(), self.clock)
-        });
-        peer.attempted();
+        self.sessions
+            .entry(id)
+            .or_insert_with(|| Session::connecting(id, persistent, self.rng.clone()))
+            .attempted();
     }
 
     pub fn connected(&mut self, remote: NodeId, link: Link) {
@@ -534,24 +534,22 @@ where
         // TODO: How should we deal with multiple peers connecting from the same IP address?
         if link.is_outbound() {
             if let Some(peer) = self.sessions.get_mut(&remote) {
-                if link.is_outbound() {
-                    self.reactor.write_all(
-                        remote,
-                        gossip::handshake(
-                            self.clock.as_secs(),
-                            &self.storage,
-                            &self.signer,
-                            self.filter.clone(),
-                            &self.config,
-                        ),
-                    );
-                }
-                peer.connected(link);
+                self.reactor.write_all(
+                    remote,
+                    gossip::handshake(
+                        self.clock.as_secs(),
+                        &self.storage,
+                        &self.signer,
+                        self.filter.clone(),
+                        &self.config,
+                    ),
+                );
+                peer.to_connected(self.clock);
             }
         } else {
             self.sessions.insert(
                 remote,
-                Session::new(
+                Session::connected(
                     remote,
                     Link::Inbound,
                     self.config.is_persistent(&remote),
@@ -568,7 +566,7 @@ where
         debug!("Disconnected from {} ({})", remote, reason);
 
         if let Some(session) = self.sessions.get_mut(&remote) {
-            session.state = session::State::Disconnected { since };
+            session.to_disconnected(since);
 
             // Attempt to re-connect to persistent peers.
             if let Some(address) = self.config.peer(&remote) {
@@ -887,6 +885,9 @@ where
                 self.reactor
                     .fetch(*remote, repo, Namespaces::default(), false);
             }
+            (session::State::Connecting { .. }, msg) => {
+                error!("Received {:?} from connecting peer {}", msg, peer.id);
+            }
             (session::State::Disconnected { .. }, msg) => {
                 debug!("Ignoring {:?} from disconnected peer {}", msg, peer.id);
             }

radicle-node/src/service/session.rs

@@ -29,6 +29,8 @@ pub enum Protocol {
 #[derive(Debug, Clone)]
 #[allow(clippy::large_enum_variant)]
 pub enum State {
+    /// Initial state for outgoing connections.
+    Connecting,
     /// Initial state after handshake protocol hand-off.
     Connected {
         /// Whether this session was initialized with a [`Message::Initialize`].
@@ -89,7 +91,20 @@ pub struct Session {
 }
 
 impl Session {
-    pub fn new(id: NodeId, link: Link, persistent: bool, rng: Rng, time: LocalTime) -> Self {
+    pub fn connecting(id: NodeId, persistent: bool, rng: Rng) -> Self {
+        Self {
+            id,
+            state: State::Connecting,
+            link: Link::Outbound,
+            subscribe: None,
+            persistent,
+            last_active: LocalTime::default(),
+            attempts: 0,
+            rng,
+        }
+    }
+
+    pub fn connected(id: NodeId, link: Link, persistent: bool, rng: Rng, time: LocalTime) -> Self {
         Self {
             id,
             state: State::Connected {
@@ -107,6 +122,10 @@ impl Session {
         }
     }
 
+    pub fn is_connecting(&self) -> bool {
+        matches!(self.state, State::Connecting { .. })
+    }
+
     pub fn is_connected(&self) -> bool {
         matches!(self.state, State::Connected { .. })
     }
@@ -134,8 +153,22 @@ impl Session {
         None
     }
 
-    pub fn connected(&mut self, _link: Link) {
+    pub fn to_connected(&mut self, since: LocalTime) {
+        assert!(
+            self.is_connecting(),
+            "Can only transition to 'connected' state from 'connecting' state"
+        );
         self.attempts = 0;
+        self.state = State::Connected {
+            initialized: false,
+            since,
+            ping: PingState::default(),
+            protocol: Protocol::default(),
+        };
+    }
+
+    pub fn to_disconnected(&mut self, since: LocalTime) {
+        self.state = State::Disconnected { since };
     }
 
     pub fn ping(&mut self, reactor: &mut Reactor) -> Result<(), Error> {