feat: Add DHT and RequestResponse notifications

.config/nextest.toml

@@ -11,12 +11,14 @@ slow-timeout = { period = "120s", terminate-after = 2 }
 fail-fast = false
 
 [test-groups]
-serial = { max-threads = 1 }
+serial = { max-threads = 4 }
 
 [[profile.default.overrides]]
 filter = 'test(/_serial$/)'
 test-group = 'serial'
+threads-required = 4
 
 [[profile.ci.overrides]]
 filter = 'test(/_serial$/)'
 test-group = 'serial'
+threads-required = 4

homestar-runtime/fixtures/defaults.toml → homestar-runtime/config/defaults.toml

@@ -50,7 +50,9 @@ mesh_n = 2
 mesh_outbound_min = 1
 
 [node.network.libp2p.dht]
-p2p_provider_timeout = 30
+p2p_receipt_timeout = 500
+p2p_workflow_info_timeout = 500
+p2p_provider_timeout = 10000
 receipt_quorum = 2
 workflow_quorum = 3
 

homestar-runtime/src/event_handler.rs

@@ -55,7 +55,9 @@ pub(crate) struct EventHandler<DB: Database> {
  receipt_quorum: usize,
  /// Minimum number of peers required to receive workflow information.
  workflow_quorum: usize,
- /// Timeout for p2p provider requests.
+ /// Timeout for p2p workflow info record requests.
+ p2p_workflow_info_timeout: Duration,
+ /// Timeout for p2p workflow info record requests from a provider.
  p2p_provider_timeout: Duration,
  /// Accessible database instance.
  db: DB,
@@ -101,7 +103,9 @@ pub(crate) struct EventHandler<DB: Database> {
  receipt_quorum: usize,
  /// Minimum number of peers required to receive workflow information.
  workflow_quorum: usize,
- /// Timeout for p2p provider requests.
+ /// Timeout for p2p workflow info record requests.
+ p2p_workflow_info_timeout: Duration,
+ /// Timeout for p2p workflow info record requests from a provider.
  p2p_provider_timeout: Duration,
  /// Accesible database instance.
  db: DB,
@@ -174,6 +178,7 @@ where
  Self {
  receipt_quorum: settings.libp2p.dht.receipt_quorum,
  workflow_quorum: settings.libp2p.dht.workflow_quorum,
+ p2p_workflow_info_timeout: settings.libp2p.dht.p2p_workflow_info_timeout,
  p2p_provider_timeout: settings.libp2p.dht.p2p_provider_timeout,
  db,
  swarm,
@@ -214,6 +219,7 @@ where
  Self {
  receipt_quorum: settings.libp2p.dht.receipt_quorum,
  workflow_quorum: settings.libp2p.dht.workflow_quorum,
+ p2p_workflow_info_timeout: settings.libp2p.dht.p2p_workflow_info_timeout,
  p2p_provider_timeout: settings.libp2p.dht.p2p_provider_timeout,
  db,
  swarm,

homestar-runtime/src/event_handler/event.rs

@@ -1,15 +1,19 @@
 //! Internal [Event] type and [Handler] implementation.
 
-use super::EventHandler;
+use super::{swarm_event::FoundEvent, EventHandler};
 #[cfg(feature = "websocket-notify")]
-use crate::event_handler::notification::{
- self, emit_receipt, EventNotificationTyp, SwarmNotification,
+use crate::event_handler::{
+ notification::{
+ self, emit_receipt, swarm::workflow_info_source_label, EventNotificationTyp,
+ SwarmNotification,
+ },
+ swarm_event::{ReceiptEvent, WorkflowInfoEvent},
 };
 #[cfg(feature = "ipfs")]
 use crate::network::IpfsCli;
 use crate::{
  db::Database,
- event_handler::{channel::AsyncChannelSender, Handler, P2PSender, ResponseEvent},
+ event_handler::{channel::AsyncChannelSender, Handler, P2PSender},
  network::{
  pubsub,
  swarm::{CapsuleTag, RequestResponseKey, TopicMessage},
@@ -32,7 +36,7 @@ use maplit::btreemap;
 use std::{collections::HashSet, num::NonZeroUsize, sync::Arc};
 #[cfg(all(feature = "ipfs", not(feature = "test-utils")))]
 use tokio::runtime::Handle;
-use tracing::{error, info, warn};
+use tracing::{debug, error, info, warn};
 
 const RENDEZVOUS_NAMESPACE: &str = "homestar";
 
@@ -105,6 +109,8 @@ pub(crate) enum Event {
  FindRecord(QueryRecord),
  /// Remove a given record from the DHT, e.g. a [Receipt].
  RemoveRecord(QueryRecord),
+ /// [Receipt] or [workflow::Info] stored event.
+ StoredRecord(FoundEvent),
  /// Outbound request event to pull data from peers.
  OutboundRequest(PeerRequest),
  /// Get providers for a record in the DHT, e.g. workflow information.
@@ -146,6 +152,49 @@ impl Event {
  }
  Event::FindRecord(record) => record.find(event_handler).await,
  Event::RemoveRecord(record) => record.remove(event_handler).await,
+ Event::StoredRecord(event) => {
+ #[allow(unused_variables)]
+ let event = event;
+
+ #[cfg(feature = "websocket-notify")]
+ match event {
+ FoundEvent::Receipt(ReceiptEvent {
+ peer_id,
+ receipt,
+ notification_type,
+ }) => notification::emit_event(
+ event_handler.ws_evt_sender(),
+ notification_type,
+ btreemap! {
+ "publisher" => peer_id.map_or(Ipld::Null, |peer_id| Ipld::String(peer_id.to_string())),
+ "cid" => Ipld::String(receipt.cid().to_string()),
+ "ran" => Ipld::String(receipt.ran().to_string())
+ },
+ ),
+ FoundEvent::Workflow(WorkflowInfoEvent {
+ peer_id,
+ workflow_info,
+ notification_type,
+ }) => {
+ if let Some(peer_label) =
+ workflow_info_source_label(notification_type.clone())
+ {
+ notification::emit_event(
+ event_handler.ws_evt_sender(),
+ notification_type,
+ btreemap! {
+ peer_label => peer_id.map_or(Ipld::Null, |peer_id| Ipld::String(peer_id.to_string())),
+ "cid" => Ipld::String(workflow_info.cid().to_string()),
+ "name" => workflow_info.name.map_or(Ipld::Null, |name| Ipld::String(name.to_string())),
+ "numTasks" => Ipld::Integer(workflow_info.num_tasks as i128),
+ "progress" => Ipld::List(workflow_info.progress.iter().map(|cid| Ipld::String(cid.to_string())).collect()),
+ "progressCount" => Ipld::Integer(workflow_info.progress_count as i128),
+ },
+ )
+ }
+ }
+ }
+ }
  Event::OutboundRequest(PeerRequest {
  peer,
  request,
@@ -171,7 +220,6 @@ impl Event {
  .map_err(anyhow::Error::new)?;
 
  let key = RequestResponseKey::new(cid.to_string().into(), capsule_tag);
-
  event_handler.query_senders.insert(query_id, (key, sender));
  }
  Event::Providers(Ok((providers, key, sender))) => {
@@ -306,8 +354,8 @@ impl Captured {
  SwarmNotification::PublishedReceiptPubsub,
  ),
  btreemap! {
- "cid" => receipt.cid().to_string(),
- "ran" => receipt.ran().to_string()
+ "cid" => Ipld::String(receipt.cid().to_string()),
+ "ran" => Ipld::String(receipt.ran().to_string())
  },
  );
  }
@@ -337,38 +385,95 @@ impl Captured {
  };
 
  if let Ok(receipt_bytes) = Receipt::invocation_capsule(&invocation_receipt) {
- let _id = event_handler
+ event_handler
  .swarm
  .behaviour_mut()
  .kademlia
  .put_record(
  Record::new(instruction_bytes, receipt_bytes.to_vec()),
  receipt_quorum,
  )
- .map_err(|err| {
- warn!(subject = "libp2p.put_record.err",
+ .map_or_else(
+ |err| {
+ warn!(subject = "libp2p.put_record.err",
  category = "publish_event",
  err=?err,
  "receipt not PUT onto DHT")
- });
+ },
+ |query_id| {
+ let key = RequestResponseKey::new(
+ receipt.cid().to_string().into(),
+ CapsuleTag::Receipt,
+ );
+ event_handler.query_senders.insert(query_id, (key, None));
+
+ debug!(
+ subject = "libp2p.put_record",
+ category = "publish_event",
+ "receipt PUT onto DHT"
+ );
+
+ #[cfg(feature = "websocket-notify")]
+ notification::emit_event(
+ event_handler.ws_evt_sender(),
+ EventNotificationTyp::SwarmNotification(
+ SwarmNotification::PutReceiptDht,
+ ),
+ btreemap! {
+ "cid" => Ipld::String(receipt.cid().to_string()),
+ "ran" => Ipld::String(receipt.ran().to_string())
+ },
+ );
+ },
+ );
 
  Arc::make_mut(&mut self.workflow).increment_progress(receipt_cid);
  let workflow_cid_bytes = self.workflow.cid_as_bytes();
  if let Ok(workflow_bytes) = self.workflow.capsule() {
- let _id = event_handler
+ event_handler
  .swarm
  .behaviour_mut()
  .kademlia
  .put_record(
  Record::new(workflow_cid_bytes, workflow_bytes),
  workflow_quorum,
  )
- .map_err(|err| {
- warn!(subject = "libp2p.put_record.err",
+ .map_or_else(
+ |err| {
+ warn!(subject = "libp2p.put_record.err",
  category = "publish_event",
  err=?err,
  "workflow information not PUT onto DHT")
- });
+ },
+ |query_id| {
+ let key = RequestResponseKey::new(
+ self.workflow.cid().to_string().into(),
+ CapsuleTag::Workflow,
+ );
+ event_handler.query_senders.insert(query_id, (key, None));
+
+ debug!(
+ subject = "libp2p.put_record",
+ category = "publish_event",
+ "workflow info PUT onto DHT"
+ );
+
+ #[cfg(feature = "websocket-notify")]
+ notification::emit_event(
+ event_handler.ws_evt_sender(),
+ EventNotificationTyp::SwarmNotification(
+ SwarmNotification::PutWorkflowInfoDht,
+ ),
+ btreemap! {
+ "cid" => Ipld::String(self.workflow.cid().to_string()),
+ "name" => self.workflow.name.as_ref().map_or(Ipld::Null, |name| Ipld::String(name.to_string())),
+ "numTasks" => Ipld::Integer(self.workflow.num_tasks as i128),
+ "progress" => Ipld::List(self.workflow.progress.iter().map(|cid| Ipld::String(cid.to_string())).collect()),
+ "progressCount" => Ipld::Integer(self.workflow.progress_count as i128),
+ },
+ )
+ },
+ );
  } else {
  error!(
  subject = "libp2p.put_record.err",
@@ -457,8 +562,8 @@ impl Replay {
  SwarmNotification::PublishedReceiptPubsub,
  ),
  btreemap! {
- "cid" => receipt.cid().to_string(),
- "ran" => receipt.ran().to_string()
+ "cid" => Ipld::String(receipt.cid().to_string()),
+ "ran" => Ipld::String(receipt.ran().to_string())
  },
  );
  })
@@ -490,14 +595,6 @@ impl QueryRecord {
  where
  DB: Database,
  {
- if event_handler.connections.peers.is_empty() {
- if let Some(sender) = self.sender {
- let _ = sender.send_async(ResponseEvent::NoPeersAvailable).await;
- }
-
- return;
- }
-
  let id = event_handler
  .swarm
  .behaviour_mut()
@@ -512,14 +609,6 @@ impl QueryRecord {
  where
  DB: Database,
  {
- if event_handler.connections.peers.is_empty() {
- if let Some(sender) = self.sender {
- let _ = sender.send_async(ResponseEvent::NoPeersAvailable).await;
- }
-
- return;
- }
-
  event_handler
  .swarm
  .behaviour_mut()
@@ -537,14 +626,6 @@ impl QueryRecord {
  where
  DB: Database,
  {
- if event_handler.connections.peers.is_empty() {
- if let Some(sender) = self.sender {
- let _ = sender.send_async(ResponseEvent::NoPeersAvailable).await;
- }
-
- return;
- }
-
  let id = event_handler
  .swarm
  .behaviour_mut()