refactor: Improve rendezvous protocol usage

.envrc

@@ -1,5 +1,5 @@
 use_flake
 
-export RUST_LOG=homestar=debug,homestar_runtime=debug,libp2p=info,libp2p_gossipsub::behaviour=debug,tarpc=info,tower_http=debug
+export RUST_LOG=homestar=debug,homestar_runtime=debug,libp2p=info,libp2p_gossipsub::behaviour=debug,tarpc=info,tower_http=debug,moka=debug
 export RUST_BACKTRACE=full
 export RUSTFLAGS="--cfg tokio_unstable"

homestar-runtime/Cargo.toml

@@ -111,6 +111,10 @@ metrics-exporter-prometheus = { version = "0.12.1", default-features = false, fe
   "http-listener",
 ], optional = true }
 miette = { version = "5.10", default-features = false, features = ["fancy"] }
+moka = { version = "0.12.1", default-features = false, features = [
+  "future",
+  "sync",
+] }
 names = { version = "0.14", default-features = false, optional = true }
 proptest = { version = "1.2", optional = true }
 puffin = { version = "0.17", default-features = false, optional = true }

homestar-runtime/fixtures/__testkey_ed25519_2.pem

@@ -0,0 +1,3 @@
+-----BEGIN PRIVATE KEY-----
+AQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQE=
+-----END PRIVATE KEY-----

homestar-runtime/fixtures/__testkey_ed25519_3.pem

@@ -0,0 +1,3 @@
+-----BEGIN PRIVATE KEY-----
+AgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgI=
+-----END PRIVATE KEY-----

homestar-runtime/fixtures/__testkey_ed25519_4.pem

@@ -0,0 +1,3 @@
+-----BEGIN PRIVATE KEY-----
+AwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwM=
+-----END PRIVATE KEY-----

homestar-runtime/fixtures/__testkey_ed25519_5.pem

@@ -0,0 +1,3 @@
+-----BEGIN PRIVATE KEY-----
+BAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQ=
+-----END PRIVATE KEY-----

homestar-runtime/src/event_handler.rs

@@ -6,7 +6,7 @@ use crate::network::ws;
 use crate::network::IpfsCli;
 use crate::{
     db::Database,
-    network::swarm::{ComposedBehaviour, RequestResponseKey},
+    network::swarm::{ComposedBehaviour, PeerDiscoveryInfo, RequestResponseKey},
     settings,
 };
 use anyhow::Result;
@@ -16,14 +16,17 @@ use libp2p::{
     core::ConnectedPoint, futures::StreamExt, kad::QueryId, rendezvous::Cookie,
     request_response::RequestId, swarm::Swarm, PeerId,
 };
+use moka::future::Cache;
 use std::{sync::Arc, time::Duration};
 use swarm_event::ResponseEvent;
-use tokio::select;
+use tokio::{runtime::Handle, select};
 
+pub(crate) mod cache;
 pub mod channel;
 pub(crate) mod error;
 pub(crate) mod event;
 pub(crate) mod swarm_event;
+pub(crate) use cache::{setup_cache, CacheValue};
 pub(crate) use error::RequestResponseError;
 pub(crate) use event::Event;
 
@@ -54,11 +57,16 @@ pub(crate) struct EventHandler<DB: Database> {
     p2p_provider_timeout: Duration,
     db: DB,
     swarm: Swarm<ComposedBehaviour>,
+    cache: Arc<Cache<String, CacheValue>>,
     sender: Arc<channel::AsyncBoundedChannelSender<Event>>,
     receiver: channel::AsyncBoundedChannelReceiver<Event>,
     query_senders: FnvHashMap<QueryId, (RequestResponseKey, Option<P2PSender>)>,
     connected_peers: FnvHashMap<PeerId, ConnectedPoint>,
+    connected_peers_limit: u32,
+    discovered_peers: FnvHashMap<PeerId, PeerDiscoveryInfo>,
     request_response_senders: FnvHashMap<RequestId, (RequestResponseKey, P2PSender)>,
+    rendezvous_registration_ttl: Duration,
+    rendezvous_discovery_interval: Duration,
     rendezvous_cookies: FnvHashMap<PeerId, Cookie>,
     pubsub_enabled: bool,
     ws_msg_sender: ws::Notifier,
@@ -76,11 +84,16 @@ pub(crate) struct EventHandler<DB: Database> {
     p2p_provider_timeout: Duration,
     db: DB,
     swarm: Swarm<ComposedBehaviour>,
+    cache: Cache<String, CacheValue>,
     sender: Arc<channel::AsyncBoundedChannelSender<Event>>,
     receiver: channel::AsyncBoundedChannelReceiver<Event>,
     query_senders: FnvHashMap<QueryId, (RequestResponseKey, Option<P2PSender>)>,
     connected_peers: FnvHashMap<PeerId, ConnectedPoint>,
+    connected_peers_limit: u32,
+    discovered_peers: FnvHashMap<PeerId, PeerDiscoveryInfo>,
     request_response_senders: FnvHashMap<RequestId, (RequestResponseKey, P2PSender)>,
+    rendezvous_registration_ttl: Duration,
+    rendezvous_discovery_interval: Duration,
     rendezvous_cookies: FnvHashMap<PeerId, Cookie>,
     pubsub_enabled: bool,
     node_addresses: Vec<libp2p::Multiaddr>,
@@ -110,17 +123,23 @@ where
         ws_msg_sender: ws::Notifier,
     ) -> Self {
         let (sender, receiver) = Self::setup_channel(settings);
+        let sender = Arc::new(sender);
         Self {
             receipt_quorum: settings.network.receipt_quorum,
             workflow_quorum: settings.network.workflow_quorum,
             p2p_provider_timeout: settings.network.p2p_provider_timeout,
             db,
             swarm,
-            sender: Arc::new(sender),
+            cache: Arc::new(setup_cache(sender.clone())),
+            sender: sender.clone(),
             receiver,
             query_senders: FnvHashMap::default(),
             request_response_senders: FnvHashMap::default(),
             connected_peers: FnvHashMap::default(),
+            connected_peers_limit: settings.network.max_connected_peers,
+            discovered_peers: FnvHashMap::default(),
+            rendezvous_registration_ttl: settings.network.rendezvous_registration_ttl,
+            rendezvous_discovery_interval: settings.network.rendezvous_discovery_interval,
             rendezvous_cookies: FnvHashMap::default(),
             pubsub_enabled: settings.network.enable_pubsub,
             ws_msg_sender,
@@ -134,17 +153,23 @@ where
     #[cfg(not(feature = "websocket-server"))]
     pub(crate) fn new(swarm: Swarm<ComposedBehaviour>, db: DB, settings: &settings::Node) -> Self {
         let (sender, receiver) = Self::setup_channel(settings);
+        let sender = Arc::new(sender);
         Self {
             receipt_quorum: settings.network.receipt_quorum,
             workflow_quorum: settings.network.workflow_quorum,
             p2p_provider_timeout: settings.network.p2p_provider_timeout,
             db,
             swarm,
-            sender: Arc::new(sender),
+            cache: Arc::new(setup_cache(sender.clone())),
+            sender: sender.clone(),
             receiver,
             query_senders: FnvHashMap::default(),
             connected_peers: FnvHashMap::default(),
+            connected_peers_limit: settings.network.max_connected_peers,
+            discovered_peers: FnvHashMap::default(),
             request_response_senders: FnvHashMap::default(),
+            rendezvous_registration_ttl: settings.network.rendezvous_registration_ttl,
+            rendezvous_discovery_interval: settings.network.rendezvous_discovery_interval,
             rendezvous_cookies: FnvHashMap::default(),
             pubsub_enabled: settings.network.enable_pubsub,
             node_addresses: settings.network.node_addresses.clone(),
@@ -189,15 +214,20 @@ where
                      swarm_event.handle_event(&mut self).await;
 
                 }
-
             }
+
+            // Poll cache for expired entries
+            self.cache.run_pending_tasks().await;
         }
     }
     /// Start [EventHandler] that matches on swarm and pubsub [events].
     ///
     /// [events]: libp2p::swarm::SwarmEvent
     #[cfg(feature = "ipfs")]
     pub(crate) async fn start(mut self, ipfs: IpfsCli) -> Result<()> {
+        let handle = Handle::current();
+        handle.spawn(poll_cache(self.cache.clone()));
+
         loop {
             select! {
                 runtime_event = self.receiver.recv_async() => {
@@ -213,3 +243,13 @@ where
         }
     }
 }
+
+/// Poll cache for expired entries
+async fn poll_cache(cache: Arc<Cache<String, CacheValue>>) {
+    let mut interval = tokio::time::interval(Duration::from_secs(1));
+
+    loop {
+        interval.tick().await;
+        cache.run_pending_tasks().await;
+    }
+}