src/eventgraph: support debugging msgs

src/event_graph/deg.rs

@@ -0,0 +1,48 @@
+/* This file is part of DarkFi (https://dark.fi)
+ *
+ * Copyright (C) 2020-2024 Dyne.org foundation
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+use crate::util::time::NanoTimestamp;
+
+macro_rules! degev {
+      ($self:expr, $event_name:ident, $($code:tt)*) => {
+          {
+              if *$self.event_graph.deg_enabled.read().await {
+                  let event = DegEvent::$event_name(deg::$event_name $($code)*);
+                  $self.event_graph.deg_notify(event).await;
+              }
+          }
+      };
+  }
+pub(crate) use degev;
+
+#[derive(Clone, Debug)]
+pub struct MessageInfo {
+    pub info: Vec<String>,
+    pub cmd: String,
+    pub time: NanoTimestamp,
+}
+
+// Needed by the degev!() macro
+pub type SendMessage = MessageInfo;
+pub type RecvMessage = MessageInfo;
+
+#[derive(Clone, Debug)]
+pub enum DegEvent {
+    SendMessage(MessageInfo),
+    RecvMessage(MessageInfo),
+}

src/event_graph/mod.rs

@@ -24,18 +24,26 @@ use std::{
 
 use async_recursion::async_recursion;
 use darkfi_serial::{deserialize_async, serialize_async};
-use log::{debug, error, info};
+use log::{debug, error, info, warn};
 use num_bigint::BigUint;
 use sled_overlay::SledTreeOverlay;
 use smol::{
     lock::{OnceCell, RwLock},
     Executor,
 };
+use tinyjson::JsonValue::{self};
 
 use crate::{
     event_graph::util::seconds_until_next_rotation,
     net::P2pPtr,
-    system::{sleep, timeout::timeout, StoppableTask, StoppableTaskPtr, Subscriber, SubscriberPtr},
+    rpc::{
+        jsonrpc::{JsonResponse, JsonResult},
+        util::json_map,
+    },
+    system::{
+        sleep, timeout::timeout, StoppableTask, StoppableTaskPtr, Subscriber, SubscriberPtr,
+        Subscription,
+    },
     Error, Result,
 };
 
@@ -51,6 +59,10 @@ use proto::{EventRep, EventReq, TipRep, TipReq, REPLY_TIMEOUT};
 mod util;
 use util::{generate_genesis, next_rotation_timestamp};
 
+// Debugging event graph
+pub(crate) mod deg;
+use deg::DegEvent;
+
 #[cfg(test)]
 mod tests;
 
@@ -95,6 +107,11 @@ pub struct EventGraph {
     days_rotation: u64,
     /// Flag signalling DAG has finished initial sync
     pub synced: RwLock<bool>,
+
+    /// Enable graph debugging
+    pub deg_enabled: RwLock<bool>,
+    /// The subscriber for which we can give dnet info over
+    deg_subscriber: SubscriberPtr<DegEvent>,
 }
 
 impl EventGraph {
@@ -124,6 +141,8 @@ impl EventGraph {
             current_genesis: RwLock::new(current_genesis.clone()),
             days_rotation,
             synced: RwLock::new(false),
+            deg_enabled: RwLock::new(false),
+            deg_subscriber: Subscriber::new(),
         });
 
         // Check if we have it in our DAG.
@@ -776,4 +795,58 @@ impl EventGraph {
 
         parents1 == parents2
     }
+
+    /// Enable graph debugging
+    pub async fn deg_enable(&self) {
+        *self.deg_enabled.write().await = true;
+        warn!("[EVENTGRAPH] Graph debugging enabled!");
+    }
+
+    /// Disable graph debugging
+    pub async fn deg_disable(&self) {
+        *self.deg_enabled.write().await = false;
+        warn!("[EVENTGRAPH] Graph debugging disabled!");
+    }
+
+    /// Subscribe to dnet events
+    pub async fn deg_subscribe(&self) -> Subscription<DegEvent> {
+        self.deg_subscriber.clone().subscribe().await
+    }
+
+    /// Send a deg notification over the subscriber
+    pub async fn deg_notify(&self, event: DegEvent) {
+        self.deg_subscriber.notify(event).await;
+    }
+
+    pub async fn eventgraph_info(&self, id: u16, _params: JsonValue) -> JsonResult {
+        let u_tips = self.unreferenced_tips.read().await.clone();
+        let u_tips_vals = u_tips
+            .into_values()
+            .map(|v| v.into_iter().map(|x| JsonValue::String(x.to_string())).collect::<Vec<_>>())
+            .collect::<Vec<_>>()
+            .concat();
+
+        let b_ids = self
+            .broadcasted_ids
+            .read()
+            .await
+            .clone()
+            .into_iter()
+            .map(|id| JsonValue::String(id.to_string()))
+            .collect::<Vec<_>>();
+
+        let values = json_map([
+            ("unreferenced_tips", JsonValue::Array(u_tips_vals)),
+            ("broadcasted_ids", JsonValue::Array(b_ids)),
+            ("synced", JsonValue::Boolean(*self.synced.read().await)),
+            (
+                "current_genesis",
+                JsonValue::String(self.current_genesis.read().await.clone().id().to_string()),
+            ),
+        ]);
+
+        let result = JsonValue::Object(HashMap::from([("eventgraph_info".to_string(), values)]));
+
+        JsonResponse::new(result, id).into()
+    }
 }

src/event_graph/proto.rs

@@ -32,7 +32,14 @@ use log::{debug, error, trace, warn};
 use smol::Executor;
 
 use super::{Event, EventGraphPtr, NULL_ID};
-use crate::{impl_p2p_message, net::*, system::timeout::timeout, Error, Result};
+use crate::{
+    event_graph::{deg, deg::degev, DegEvent},
+    impl_p2p_message,
+    net::*,
+    system::timeout::timeout,
+    util::time::NanoTimestamp,
+    Error, Result,
+};
 
 /// Malicious behaviour threshold. If the threshold is reached, we will
 /// drop the peer from our P2P connection.
@@ -174,6 +181,12 @@ impl ProtocolEventGraph {
                 continue
             }
 
+            degev!(self, RecvMessage, {
+                info: vec![event_id.to_string()],
+                cmd: "EventPut".to_string(),
+                time: NanoTimestamp::current_time(),
+            });
+
             // We received an event. Check if we already have it in our DAG.
             // Check event is not older that current genesis event timestamp.
             // Also check if we have the event's parents. In the case we do
@@ -341,6 +354,12 @@ impl ProtocolEventGraph {
                 continue
             }
 
+            degev!(self, SendMessage, {
+                info: vec![event_id.to_string()],
+                cmd: "EventPut".to_string(),
+                time: NanoTimestamp::current_time(),
+            });
+
             // Relay the event to other peers.
             self.event_graph
                 .p2p
@@ -358,8 +377,8 @@ impl ProtocolEventGraph {
                 Err(_) => continue,
             };
             trace!(
-                target: "event_graph::protocol::handle_multi_event_req()",
-                "Got MultiEventReq: {:?} [{}]", event_ids, self.channel.address(),
+                target: "event_graph::protocol::handle_event_req()",
+                "Got EventReq: {:?} [{}]", event_ids, self.channel.address(),
             );
 
             // Check if node has finished syncing its DAG
@@ -371,6 +390,14 @@ impl ProtocolEventGraph {
                 continue
             }
 
+            let info = event_ids.clone().into_iter().map(|x| x.to_string()).collect();
+
+            degev!(self, RecvMessage, {
+                info,
+                cmd: "EventReq".to_string(),
+                time: NanoTimestamp::current_time(),
+            });
+
             // We received an event request from somebody.
             // If we do have it, we will send it back to them as `EventRep`.
             // Otherwise, we'll stay quiet. An honest node should always have
@@ -383,8 +410,8 @@ impl ProtocolEventGraph {
             // against malicious event requests where they want us to keep
             // reading our db and steal our bandwidth.
             let mut events = vec![];
-            for event_id in event_ids {
-                if !self.event_graph.broadcasted_ids.read().await.contains(&event_id) {
+            for event_id in event_ids.iter() {
+                if !self.event_graph.broadcasted_ids.read().await.contains(event_id) {
                     let malicious_count = self.malicious_count.fetch_add(1, SeqCst);
                     if malicious_count + 1 == MALICIOUS_THRESHOLD {
                         error!(
@@ -410,7 +437,7 @@ impl ProtocolEventGraph {
                     target: "event_graph::protocol::handle_event_req()",
                     "Fetching event {:?} from DAG", event_id,
                 );
-                events.push(self.event_graph.dag_get(&event_id).await.unwrap().unwrap());
+                events.push(self.event_graph.dag_get(event_id).await.unwrap().unwrap());
             }
 
             // Check if the incoming event is older than the genesis event. If so, something
@@ -442,6 +469,14 @@ impl ProtocolEventGraph {
             //bcast_ids.remove(&event_id);
             drop(bcast_ids);
 
+            let info = event_ids.into_iter().map(|x| x.to_string()).collect();
+
+            degev!(self, SendMessage, {
+                info,
+                cmd: "EventRep".to_string(),
+                time: NanoTimestamp::current_time(),
+            });
+
             // Reply with the event
             self.channel.send(&EventRep(events)).await?;
         }
@@ -453,6 +488,11 @@ impl ProtocolEventGraph {
     async fn handle_tip_req(self: Arc<Self>) -> Result<()> {
         loop {
             self.tip_req_sub.receive().await?;
+            degev!(self, RecvMessage, {
+                info: vec![],
+                cmd: "TipReq".to_string(),
+                time: NanoTimestamp::current_time(),
+            });
             trace!(
                 target: "event_graph::protocol::handle_tip_req()",
                 "Got TipReq [{}]", self.channel.address(),
@@ -480,6 +520,21 @@ impl ProtocolEventGraph {
             }
             drop(bcast_ids);
 
+            let info = layers
+                .clone()
+                .into_values()
+                .map(|v| v.into_iter().map(|id| id.to_string()).collect::<Vec<_>>())
+                .collect::<Vec<_>>()
+                .concat();
+
+            degev!(self, SendMessage, {
+                info,
+                cmd: "TipRep".to_string(),
+                time: NanoTimestamp::current_time(),
+            });
+
+            let _ = self.event_graph.eventgraph_info(1, tinyjson::JsonValue::Array(vec![])).await;
+
             self.channel.send(&TipRep(layers)).await?;
         }
     }