feat: Add rendezvous rediscovery and registration renewal

Add cache with eviction listener to trigger discovery and renewal events
on an interval and at TTL respectively.

.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/src/event_handler.rs

@@ -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;
 
+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,6 +57,7 @@ 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>)>,
@@ -80,6 +84,7 @@ 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>)>,
@@ -118,13 +123,15 @@ 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: setup_cache(sender.clone()),
+ sender: sender.clone(),
  receiver,
  query_senders: FnvHashMap::default(),
  request_response_senders: FnvHashMap::default(),
@@ -146,13 +153,15 @@ 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: setup_cache(sender.clone()),
+ sender: sender.clone(),
  receiver,
  query_senders: FnvHashMap::default(),
  connected_peers: FnvHashMap::default(),
@@ -205,8 +214,10 @@ 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].
@@ -226,6 +237,9 @@ where
  swarm_event.handle_event(&mut self, ipfs_clone).await;
  }
  }
+
+ // Poll cache for expired entries
+ self.cache.run_pending_tasks().await;
  }
  }
 }

homestar-runtime/src/event_handler/cache.rs

@@ -0,0 +1,91 @@
+use crate::{channel, event_handler::Event};
+use libp2p::PeerId;
+use moka::{future::Cache, Expiry as ExpiryBase};
+use std::{
+ collections::HashMap,
+ sync::Arc,
+ time::{Duration, Instant},
+};
+
+struct Expiry;
+
+impl ExpiryBase<String, CacheValue> for Expiry {
+ fn expire_after_create(
+ &self,
+ _key: &String,
+ value: &CacheValue,
+ _current_time: Instant,
+ ) -> Option<Duration> {
+ value.expiration.as_duration()
+ }
+}
+
+#[derive(Clone, Debug)]
+pub(crate) struct CacheValue {
+ expiration: Expiration,
+ data: HashMap<String, CacheData>,
+}
+
+impl CacheValue {
+ pub(crate) fn new(expiration: Expiration, data: HashMap<String, CacheData>) -> Self {
+ Self { expiration, data }
+ }
+}
+
+#[derive(Clone, Debug)]
+pub(crate) enum CacheData {
+ Peer(PeerId),
+ OnEviction(DispatchEvent),
+}
+
+#[derive(Clone, Debug)]
+pub(crate) enum DispatchEvent {
+ RegisterPeer,
+ DiscoverPeers,
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub(crate) enum Expiration {
+ Registration(Duration),
+ Discovery(Duration),
+}
+
+impl Expiration {
+ pub(crate) fn as_duration(&self) -> Option<Duration> {
+ match self {
+ Expiration::Registration(ttl) => Some(*ttl),
+ Expiration::Discovery(interval) => Some(*interval),
+ }
+ }
+}
+
+pub(crate) fn setup_cache(
+ sender: Arc<channel::AsyncBoundedChannelSender<Event>>,
+) -> Cache<String, CacheValue> {
+ let eviction_listener = move |_key: Arc<String>, val: CacheValue, _cause| {
+ let tx = Arc::clone(&sender);
+
+ if let Some(CacheData::OnEviction(event)) = val.data.get("on_eviction") {
+ match event {
+ DispatchEvent::RegisterPeer => {
+ if let Some(CacheData::Peer(rendezvous_node)) = val.data.get("rendezvous_node")
+ {
+ let _ = tx.send(Event::RegisterPeer(rendezvous_node.to_owned()));
+ };
+ }
+ DispatchEvent::DiscoverPeers => {
+ if let Some(CacheData::Peer(rendezvous_node)) = val.data.get("rendezvous_node")
+ {
+ let _ = tx.send(Event::DiscoverPeers(rendezvous_node.to_owned()));
+ };
+ }
+ }
+ }
+ };
+
+ Cache::builder()
+ .expire_after(Expiry)
+ .time_to_live(Duration::from_secs(5))
+ .eviction_listener(eviction_listener)
+ .build()
+}