Move relay connection logic into main event loop (#632)

* Bump `libp2p` to version `0.53.2` (#631)

* Bump libp2p to version 0.53.2

* We don't need to listen on tcp port when in relay mode

* Listening on relay circuit no longer sometimes fails

* Remove tcp feature requirement from libp2p

* Refactor connection_keep_alive method

* Clippy

* Remove unnecessary connection_keep_alive method from peers behaviour

* Add CHANGELOG.md entry

---------

Co-authored-by: adz <x12@adz.garden>

* Move network service relay initialization into main event loop

* Clippy

* Add DCUTR event debug logging to swarm

* Change log message

* Adjust connection limits

* Even nicer log messages

* Helper to print or info log depending on log level

* Listening on relay circuit no longer sometimes fails

---------

Co-authored-by: adz <x12@adz.garden>

CHANGELOG.md

@@ -9,7 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
-- Upate `libp2p` to version `0.53.2` and apply API changes [#631](https://github.com/p2panda/aquadoggo/pull/631)
+- Update `libp2p` to version `0.53.2` and apply API changes [#631](https://github.com/p2panda/aquadoggo/pull/631)
+- Move relay registration into main network service event loop [#632](https://github.com/p2panda/aquadoggo/pull/632)
 
 ### Fixed
 

aquadoggo/src/http/service.rs

@@ -18,6 +18,7 @@ use crate::http::api::{
     handle_blob_document, handle_blob_view, handle_graphql_playground, handle_graphql_query,
 };
 use crate::http::context::HttpServiceContext;
+use crate::info_or_print;
 use crate::manager::{ServiceReadySender, Shutdown};
 
 /// Route to the GraphQL playground
@@ -74,17 +75,19 @@ pub async fn http_service(
     let builder = if let Ok(builder) = axum::Server::try_bind(&http_address) {
         builder
     } else {
-        println!("HTTP port {http_port} was already taken, try random port instead ..");
+        info_or_print(&format!(
+            "HTTP port {http_port} was already taken, try random port instead .."
+        ));
         axum::Server::try_bind(&SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), 0))?
     };
 
     let builder = builder.serve(build_server(http_context).into_make_service());
 
     let local_address = builder.local_addr();
-    println!(
+    info_or_print(&format!(
         "Go to http://{}/graphql to use GraphQL playground",
         local_address
-    );
+    ));
 
     builder
         .with_graceful_shutdown(async {

aquadoggo/src/lib.rs

@@ -32,6 +32,8 @@ mod test_utils;
 #[cfg(test)]
 mod tests;
 
+use log::{info, log_enabled, Level};
+
 pub use crate::api::{ConfigFile, LockFile};
 pub use crate::config::{AllowList, Configuration};
 pub use crate::network::NetworkConfiguration;
@@ -59,3 +61,14 @@ fn init() {
         let _ = env_logger::builder().is_test(true).try_init();
     }
 }
+
+/// Helper method for logging a message directly to standard out or via the `log` crate when any
+/// logging level is enabled. We need this as some messages should be always printed, but when any
+/// logging level is selected, we want the message to be printed with consistent formatting.
+fn info_or_print(message: &str) {
+    if log_enabled!(Level::Info) || log_enabled!(Level::Debug) || log_enabled!(Level::Trace) {
+        info!("{message}");
+    } else {
+        println!("{message}");
+    }
+}

aquadoggo/src/network/behaviour.rs

@@ -223,7 +223,6 @@ pub enum Event {
     RendezvousClient(rendezvous::client::Event),
     #[allow(dead_code)]
     RendezvousServer(rendezvous::server::Event),
-    #[allow(dead_code)]
     Dcutr(dcutr::Event),
     Peers(peers::Event),
     Void,

aquadoggo/src/network/config.rs

@@ -132,9 +132,9 @@ impl Default for NetworkConfiguration {
             dial_concurrency_factor: 8,
             max_connections_in: 16,
             max_connections_out: 16,
-            max_connections_pending_in: 16,
-            max_connections_pending_out: 16,
-            max_connections_per_peer: 8,
+            max_connections_pending_in: 8,
+            max_connections_pending_out: 8,
+            max_connections_per_peer: 2,
         }
     }
 }

aquadoggo/src/network/mod.rs

@@ -4,6 +4,7 @@ mod behaviour;
 mod config;
 pub mod identity;
 mod peers;
+mod relay;
 mod service;
 mod shutdown;
 mod swarm;

aquadoggo/src/network/peers/behaviour.rs

@@ -120,15 +120,6 @@ impl Behaviour {
         false
     }
 
-    /// Disable the behaviour, it won't handle any connection events or received messages.
-    pub fn disable(&mut self) {
-        self.enabled = false
-    }
-
-    pub fn enable(&mut self) {
-        self.enabled = true
-    }
-
     pub fn send_message(&mut self, peer: Peer, message: PeerMessage) {
         self.push_event(ToSwarm::NotifyHandler {
             peer_id: peer.id(),

aquadoggo/src/network/relay.rs

@@ -0,0 +1,110 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+use libp2p::multiaddr::Protocol;
+use libp2p::{rendezvous, Multiaddr, PeerId, Swarm};
+
+use crate::network::behaviour::P2pandaBehaviour;
+use crate::network::config::NODE_NAMESPACE;
+
+/// A relay node.
+pub struct Relay {
+    /// PeerId of the relay node.
+    pub(crate) peer_id: PeerId,
+
+    /// A single Multiaddr which we know the relay node to be accessible at.
+    pub(crate) addr: Multiaddr,
+
+    /// The namespace we discover peers at on this relay.
+    pub(crate) namespace: String,
+
+    /// Did we tell the relay it's observed address yet.
+    pub(crate) told_addr: bool,
+
+    /// Are we currently discovering peers.
+    pub(crate) discovering: bool,
+
+    /// Are we in the process of registering at this relay.
+    pub(crate) registering: bool,
+
+    /// Have we successfully registered.
+    pub(crate) registered: bool,
+
+    /// Was our relay circuit reservation accepted.
+    pub(crate) reservation_accepted: bool,
+}
+
+impl Relay {
+    pub fn new(peer_id: PeerId, addr: Multiaddr) -> Self {
+        Relay {
+            peer_id,
+            addr,
+            namespace: NODE_NAMESPACE.to_string(),
+            told_addr: false,
+            discovering: false,
+            registering: false,
+            registered: false,
+            reservation_accepted: false,
+        }
+    }
+
+    /// The circuit address we should listen at for this relay.
+    pub fn circuit_addr(&self) -> Multiaddr {
+        self.addr
+            .clone()
+            .with(Protocol::P2p(self.peer_id))
+            .with(Protocol::P2pCircuit)
+    }
+
+    /// Start listening on the relay circuit address and register on our discovery namespace.
+    pub fn register(&mut self, swarm: &mut Swarm<P2pandaBehaviour>) -> Result<bool, anyhow::Error> {
+        if self.registered || self.registering {
+            return Ok(false);
+        }
+
+        self.registering = true;
+
+        // Start listening on the circuit relay address.
+        let circuit_address = self.circuit_addr();
+        swarm.listen_on(circuit_address.clone())?;
+
+        // Register in the `NODE_NAMESPACE` using the rendezvous network behaviour.
+        swarm
+            .behaviour_mut()
+            .rendezvous_client
+            .as_mut()
+            .unwrap()
+            .register(
+                rendezvous::Namespace::from_static(NODE_NAMESPACE),
+                self.peer_id,
+                None, // Default ttl is 7200s
+            )?;
+
+        Ok(true)
+    }
+
+    /// Start discovering peers also registered at the same namespace.
+    pub fn discover(&mut self, swarm: &mut Swarm<P2pandaBehaviour>) -> bool {
+        if self.reservation_accepted && self.registered && !self.discovering {
+            self.discovering = true;
+
+            swarm
+                .behaviour_mut()
+                .rendezvous_client
+                .as_mut()
+                .expect("Relay client behaviour exists")
+                .discover(
+                    Some(
+                        rendezvous::Namespace::new(NODE_NAMESPACE.to_string())
+                            .expect("Valid namespace"),
+                    ),
+                    None,
+                    None,
+                    self.peer_id,
+                );
+
+            true
+        } else {
+            false
+        }
+    }
+}