net: Remove most of dnet and replace with a ChannelInfo subscriber.

We want to make dnet event-based and not a polling thing, and we also
want to remove the JSON dependency directly in the p2p library. Such
format conversions should happen out of the library, likely in the
daemon providing debugging info.

src/net/channel.rs

@@ -37,7 +37,7 @@ use super::{
 };
 use crate::{
     system::{StoppableTask, StoppableTaskPtr, Subscriber, SubscriberPtr, Subscription},
-    util::{ringbuffer::RingBuffer, time::NanoTimestamp},
+    util::time::NanoTimestamp,
     Error, Result,
 };
 
@@ -45,24 +45,26 @@ use crate::{
 pub type ChannelPtr = Arc<Channel>;
 
 /// Channel debug info
-#[derive(Clone)]
+#[derive(Clone, Debug)]
 pub struct ChannelInfo {
     pub addr: Url,
-    pub random_id: u32,
+    pub random_id: usize,
     pub remote_node_id: String,
-    pub log: RingBuffer<(NanoTimestamp, String, String), 512>,
+    pub time: NanoTimestamp,
+    pub op: String,
+    pub cmd: String,
 }
 
 impl ChannelInfo {
     fn new(addr: Url) -> Self {
-        Self { addr, random_id: OsRng.gen(), remote_node_id: String::new(), log: RingBuffer::new() }
-    }
-
-    /// Get available debug info, resets the ringbuffer when called.
-    fn dnet_info(&mut self) -> Self {
-        let info = self.clone();
-        self.log = RingBuffer::new();
-        info
+        Self {
+            addr,
+            random_id: OsRng.gen(),
+            remote_node_id: String::new(),
+            time: NanoTimestamp::current_time(),
+            op: String::new(),
+            cmd: String::new(),
+        }
     }
 }
 
@@ -85,7 +87,7 @@ pub struct Channel {
     /// Weak pointer to respective session
     session: SessionWeakPtr,
     /// Channel debug info
-    info: Mutex<Option<ChannelInfo>>,
+    info: Mutex<ChannelInfo>,
 }
 
 impl std::fmt::Debug for Channel {
@@ -110,11 +112,7 @@ impl Channel {
         let message_subsystem = MessageSubsystem::new();
         Self::setup_dispatchers(&message_subsystem).await;
 
-        let info = if *session.upgrade().unwrap().p2p().dnet_enabled.lock().await {
-            Mutex::new(Some(ChannelInfo::new(address.clone())))
-        } else {
-            Mutex::new(None)
-        };
+        let info = Mutex::new(ChannelInfo::new(address.clone()));
 
         Arc::new(Self {
             reader,
@@ -139,26 +137,6 @@ impl Channel {
         subsystem.add_dispatch::<message::AddrsMessage>().await;
     }
 
-    /// Fetch dnet info for the channel, if enabled.
-    /// Returns the [`Channel::address`] and [`ChannelInfo`].
-    pub(crate) async fn dnet_info(&self) -> ChannelInfo {
-        // We're unwrapping here because if we get None it means
-        // there's a bug somehwere where we initialized dnet but
-        // ChannelInfo was not created.
-        self.info.lock().await.as_mut().unwrap().dnet_info()
-    }
-
-    pub(crate) async fn dnet_enable(&self) {
-        let mut info = self.info.lock().await;
-        if info.is_none() {
-            *info = Some(ChannelInfo::new(self.address.clone()));
-        }
-    }
-
-    pub(crate) async fn dnet_disable(&self) {
-        *self.info.lock().await = None;
-    }
-
     /// Starts the channel. Runs a receive loop to start receiving messages
     /// or handles a network failure.
     pub fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) {
@@ -247,11 +225,11 @@ impl Channel {
         let packet = Packet { command: M::NAME.to_string(), payload: serialize(message) };
 
         dnet!(self,
-            let time = NanoTimestamp::current_time();
-            match self.info.lock().await.as_mut() {
-                Some(info) => info.log.push((time, "send".into(), packet.command.clone())),
-                None => unreachable!(),
-            }
+            let mut info = self.info.lock().await;
+            info.time = NanoTimestamp::current_time();
+            info.op = "send".to_string();
+            info.cmd = packet.command.clone();
+            self.p2p().dnet_sub().notify(info.clone()).await;
         );
 
         let stream = &mut *self.writer.lock().await;

src/net/p2p.rs

@@ -29,15 +29,13 @@ use smol::Executor;
 use url::Url;
 
 use super::{
-    channel::ChannelPtr,
+    channel::{ChannelInfo, ChannelPtr},
     hosts::{Hosts, HostsPtr},
     message::Message,
     protocol::{protocol_registry::ProtocolRegistry, register_default_protocols},
     session::{
-        inbound_session::InboundDnet,
-        outbound_session::{OutboundDnet, OutboundState},
         InboundSession, InboundSessionPtr, ManualSession, ManualSessionPtr, OutboundSession,
-        OutboundSessionPtr, SeedSyncSession, Session,
+        OutboundSessionPtr, SeedSyncSession,
     },
     settings::{Settings, SettingsPtr},
 };
@@ -53,48 +51,6 @@ pub type ConnectedChannels = Mutex<HashMap<Url, ChannelPtr>>;
 /// Atomic pointer to the p2p interface
 pub type P2pPtr = Arc<P2p>;
 
-/// Representations of the p2p state
-enum P2pState {
-    /// The P2P object has been created but not yet started
-    Open,
-    /// We are performing the initial seed session
-    Start,
-    /// Seed session finished, but not yet running
-    Started,
-    /// P2P is running and the network is active
-    Run,
-    /// The P2P network has been stopped
-    Stopped,
-}
-
-/// Types of DnetInfo (used with sessions)
-pub enum DnetInfo {
-    /// Hosts info
-    Hosts(Vec<Url>),
-    /// Outbound Session Info
-    Outbound(OutboundDnet),
-    /// Inbound Session Info
-    Inbound(InboundDnet),
-    // Manual Session Info
-    //Manual(ManualDnet),
-}
-
-impl std::fmt::Display for P2pState {
-    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
-        write!(
-            f,
-            "{}",
-            match self {
-                Self::Open => "open",
-                Self::Start => "start",
-                Self::Started => "started",
-                Self::Run => "run",
-                Self::Stopped => "stopped",
-            }
-        )
-    }
-}
-
 /// Toplevel peer-to-peer networking interface
 pub struct P2p {
     /// Channels pending connection
@@ -109,8 +65,6 @@ pub struct P2p {
     hosts: HostsPtr,
     /// Protocol registry
     protocol_registry: ProtocolRegistry,
-    /// The state of the interface
-    state: Mutex<P2pState>,
     /// P2P network settings
     settings: SettingsPtr,
     /// Boolean lock marking if peer discovery is active
@@ -125,6 +79,8 @@ pub struct P2p {
 
     /// Enable network debugging
     pub dnet_enabled: Mutex<bool>,
+    /// The subscriber for which we can give dnet info over
+    dnet_sub: SubscriberPtr<ChannelInfo>,
 }
 
 impl P2p {
@@ -146,7 +102,6 @@ impl P2p {
             stop_subscriber: Subscriber::new(),
             hosts: Hosts::new(settings.clone()),
             protocol_registry: ProtocolRegistry::new(),
-            state: Mutex::new(P2pState::Open),
             settings,
             peer_discovery_running: Mutex::new(false),
 
@@ -155,6 +110,7 @@ impl P2p {
             session_outbound: Mutex::new(None),
 
             dnet_enabled: Mutex::new(false),
+            dnet_sub: Subscriber::new(),
         });
 
         let parent = Arc::downgrade(&self_);
@@ -172,15 +128,12 @@ impl P2p {
     pub async fn start(self: Arc<Self>, ex: Arc<Executor<'_>>) -> Result<()> {
         debug!(target: "net::p2p::start()", "P2P::start() [BEGIN]");
         info!(target: "net::p2p::start()", "[P2P] Seeding P2P subsystem");
-        *self.state.lock().await = P2pState::Start;
 
         // Start seed session
         let seed = SeedSyncSession::new(Arc::downgrade(&self));
         // This will block until all seed queries have finished
         seed.start(ex.clone()).await?;
 
-        *self.state.lock().await = P2pState::Started;
-
         debug!(target: "net::p2p::start()", "P2P::start() [END]");
         Ok(())
     }
@@ -204,7 +157,6 @@ impl P2p {
     pub async fn run(self: Arc<Self>, ex: Arc<Executor<'_>>) -> Result<()> {
         debug!(target: "net::p2p::run()", "P2P::run() [BEGIN]");
         info!(target: "net::p2p::run()", "[P2P] Running P2P subsystem");
-        *self.state.lock().await = P2pState::Run;
 
         // First attempt any set manual connections
         let manual = self.session_manual().await;
@@ -233,8 +185,6 @@ impl P2p {
         inbound.stop().await;
         outbound.stop().await;
 
-        *self.state.lock().await = P2pState::Stopped;
-
         debug!(target: "net::p2p::run()", "P2P::run() [END]");
         Ok(())
     }
@@ -358,11 +308,6 @@ impl P2p {
 
     /// Enable network debugging
     pub async fn dnet_enable(&self) {
-        // Enable log for all connected channels if not enabled already
-        for channel in self.channels().lock().await.values() {
-            channel.dnet_enable().await;
-        }
-
         *self.dnet_enabled.lock().await = true;
         warn!("[P2P] Network debugging enabled!");
     }
@@ -370,101 +315,12 @@ impl P2p {
     /// Disable network debugging
     pub async fn dnet_disable(&self) {
         *self.dnet_enabled.lock().await = false;
-
-        // Clear out any held data
-        for channel in self.channels().lock().await.values() {
-            channel.dnet_disable().await;
-        }
-
         warn!("[P2P] Network debugging disabled!");
     }
 
-    /// Gather session dnet info and return it in a vec.
-    /// Returns an empty vec if dnet is disabled.
-    pub async fn dnet_info(&self) -> Vec<DnetInfo> {
-        let mut ret = vec![];
-
-        if *self.dnet_enabled.lock().await {
-            ret.push(self.session_inbound().await.dnet_info().await);
-            ret.push(self.session_outbound().await.dnet_info().await);
-            ret.push(DnetInfo::Hosts(self.hosts.load_all().await));
-        }
-
-        ret
-    }
-
-    /// Maps DnetInfo into a JSON struct usable by clients
-    pub fn map_dnet_info(dnet_info: Vec<DnetInfo>) -> serde_json::Value {
-        let mut map = serde_json::Map::new();
-        map.insert("inbound".into(), serde_json::Value::Null);
-        map.insert("outbound".into(), serde_json::Value::Null);
-        map.insert("hosts".into(), serde_json::Value::Null);
-
-        // We assume there will be one of each
-        for info in dnet_info {
-            match info {
-                DnetInfo::Hosts(hosts) => map["hosts"] = serde_json::json!(hosts),
-
-                DnetInfo::Outbound(outbound_info) => {
-                    let mut slot_info = vec![];
-                    for slot in outbound_info.slots {
-                        let Some(slot) = slot else {
-                            slot_info.push(serde_json::Value::Null);
-                            continue
-                        };
-
-                        let obj = if slot.state != OutboundState::Open {
-                            serde_json::json!({
-                                "addr": slot.addr.unwrap().to_string(),
-                                "state": slot.state.to_string(),
-                                "info": {
-                                    "addr": slot.channel.as_ref().unwrap().addr.to_string(),
-                                    "random_id": slot.channel.as_ref().unwrap().random_id,
-                                    "remote_id": slot.channel.as_ref().unwrap().remote_node_id,
-                                    "log": slot.channel.as_ref().unwrap().log.to_vec(),
-                                }
-                            })
-                        } else {
-                            serde_json::json!({
-                                "addr": serde_json::Value::Null,
-                                "state": slot.state.to_string(),
-                                "info": serde_json::Value::Null,
-                            })
-                        };
-
-                        slot_info.push(obj);
-                    }
-
-                    map["outbound"] = serde_json::json!(slot_info);
-                }
-
-                DnetInfo::Inbound(inbound_info) => {
-                    let mut slot_info = vec![];
-                    for slot in inbound_info.slots {
-                        let Some(slot) = slot else {
-                            slot_info.push(serde_json::Value::Null);
-                            continue
-                        };
-
-                        let obj = serde_json::json!({
-                            "addr": slot.addr.unwrap().to_string(),
-                            "info": {
-                                "addr": slot.channel.as_ref().unwrap().addr.to_string(),
-                                "random_id": slot.channel.as_ref().unwrap().random_id,
-                                "remote_id": slot.channel.as_ref().unwrap().remote_node_id,
-                                "log": slot.channel.as_ref().unwrap().log.to_vec(),
-                            }
-                        });
-
-                        slot_info.push(obj);
-                    }
-
-                    map["inbound"] = serde_json::json!(slot_info);
-                }
-            }
-        }
-
-        serde_json::json!(map)
+    /// Return a reference to the dnet subscriber
+    pub fn dnet_sub(&self) -> SubscriberPtr<ChannelInfo> {
+        self.dnet_sub.clone()
     }
 }
 

src/net/session/inbound_session.rs

@@ -23,8 +23,6 @@
 //! an acceptor pointer, and a stoppable task pointer. Using a weak pointer
 //! to P2P allows us to avoid circular dependencies.
 
-use std::collections::HashMap;
-
 use async_std::sync::{Arc, Mutex, Weak};
 use async_trait::async_trait;
 use log::{debug, error, info};
@@ -34,8 +32,8 @@ use url::Url;
 use super::{
     super::{
         acceptor::{Acceptor, AcceptorPtr},
-        channel::{ChannelInfo, ChannelPtr},
-        p2p::{DnetInfo, P2p, P2pPtr},
+        channel::ChannelPtr,
+        p2p::{P2p, P2pPtr},
     },
     Session, SessionBitFlag, SESSION_INBOUND,
 };
@@ -46,40 +44,17 @@ use crate::{
 
 pub type InboundSessionPtr = Arc<InboundSession>;
 
-/// dnet info for an inbound connection
-#[derive(Clone)]
-pub struct InboundInfo {
-    /// Remote address
-    pub addr: Option<Url>,
-    /// Channel info
-    pub channel: Option<ChannelInfo>,
-}
-
-impl InboundInfo {
-    async fn dnet_info(&self, p2p: P2pPtr) -> Option<Self> {
-        let addr = self.addr.clone()?;
-        let chan = p2p.channels().lock().await.get(&addr).cloned()?;
-        Some(Self { addr: Some(addr), channel: Some(chan.dnet_info().await) })
-    }
-}
-
 /// Defines inbound connections session
 pub struct InboundSession {
     p2p: Weak<P2p>,
     acceptors: Mutex<Vec<AcceptorPtr>>,
     accept_tasks: Mutex<Vec<StoppableTaskPtr>>,
-    connect_infos: Mutex<Vec<HashMap<Url, InboundInfo>>>,
 }
 
 impl InboundSession {
     /// Create a new inbound session
     pub fn new(p2p: Weak<P2p>) -> InboundSessionPtr {
-        Arc::new(Self {
-            p2p,
-            acceptors: Mutex::new(vec![]),
-            accept_tasks: Mutex::new(vec![]),
-            connect_infos: Mutex::new(vec![]),
-        })
+        Arc::new(Self { p2p, acceptors: Mutex::new(vec![]), accept_tasks: Mutex::new(vec![]) })
     }
 
     /// Starts the inbound session. Begins by accepting connections and fails
@@ -107,7 +82,6 @@ impl InboundSession {
                 ex.clone(),
             );
 
-            self.connect_infos.lock().await.push(HashMap::new());
             accept_tasks.push(task);
         }
 
@@ -180,30 +154,20 @@ impl InboundSession {
         let stop_sub = channel.subscribe_stop().await?;
 
         self.register_channel(channel.clone(), ex.clone()).await?;
-        let addr = channel.address().clone();
-
-        self.connect_infos.lock().await[index]
-            .insert(addr.clone(), InboundInfo { addr: Some(addr.clone()), channel: None });
 
         stop_sub.receive().await;
+
         debug!(
             target: "net::inbound_session::setup_channel()",
             "Received stop_sub, removing channel from P2P",
         );
 
         self.p2p().remove(channel).await;
-        self.connect_infos.lock().await[index].remove(&addr);
 
         Ok(())
     }
 }
 
-/// Dnet information for the inbound session
-pub struct InboundDnet {
-    /// Slot information
-    pub slots: Vec<Option<InboundInfo>>,
-}
-
 #[async_trait]
 impl Session for InboundSession {
     fn p2p(&self) -> P2pPtr {
@@ -213,16 +177,4 @@ impl Session for InboundSession {
     fn type_id(&self) -> SessionBitFlag {
         SESSION_INBOUND
     }
-
-    async fn dnet_info(&self) -> DnetInfo {
-        let mut slots = vec![];
-
-        for listen_addr in (*self.connect_infos.lock().await).iter() {
-            for slot in listen_addr.values() {
-                slots.push(slot.dnet_info(self.p2p()).await);
-            }
-        }
-
-        DnetInfo::Inbound(InboundDnet { slots })
-    }
 }

src/net/session/manual_session.rs

@@ -39,7 +39,7 @@ use super::{
     super::{
         channel::ChannelPtr,
         connector::Connector,
-        p2p::{DnetInfo, P2p, P2pPtr},
+        p2p::{P2p, P2pPtr},
     },
     Session, SessionBitFlag, SESSION_MANUAL,
 };
@@ -212,8 +212,4 @@ impl Session for ManualSession {
     fn type_id(&self) -> SessionBitFlag {
         SESSION_MANUAL
     }
-
-    async fn dnet_info(&self) -> DnetInfo {
-        todo!()
-    }
 }

src/net/session/mod.rs

@@ -21,11 +21,7 @@ use async_trait::async_trait;
 use log::debug;
 use smol::Executor;
 
-use super::{
-    channel::ChannelPtr,
-    p2p::{DnetInfo, P2pPtr},
-    protocol::ProtocolVersion,
-};
+use super::{channel::ChannelPtr, p2p::P2pPtr, protocol::ProtocolVersion};
 use crate::Result;
 
 pub mod inbound_session;
@@ -154,7 +150,4 @@ pub trait Session: Sync {
 
     /// Return the session bit flag for the session type
     fn type_id(&self) -> SessionBitFlag;
-
-    /// Get network debug info
-    async fn dnet_info(&self) -> DnetInfo;
 }

src/net/session/outbound_session.rs

@@ -36,10 +36,10 @@ use url::Url;
 
 use super::{
     super::{
-        channel::{ChannelInfo, ChannelPtr},
+        channel::ChannelPtr,
         connector::Connector,
         message::GetAddrsMessage,
-        p2p::{DnetInfo, P2p, P2pPtr},
+        p2p::{P2p, P2pPtr},
     },
     Session, SessionBitFlag, SESSION_OUTBOUND,
 };
@@ -52,7 +52,7 @@ use crate::{
 pub type OutboundSessionPtr = Arc<OutboundSession>;
 
 /// Connection state
-#[derive(Eq, PartialEq, Copy, Clone)]
+#[derive(Eq, PartialEq, Copy, Clone, Debug)]
 pub enum OutboundState {
     Open,
     Pending,
@@ -73,31 +73,6 @@ impl std::fmt::Display for OutboundState {
     }
 }
 
-/// dnet info for an outbound connection
-#[derive(Clone)]
-pub struct OutboundInfo {
-    /// Remote address
-    pub addr: Option<Url>,
-    /// Channel info
-    pub channel: Option<ChannelInfo>,
-    /// Connection state
-    pub state: OutboundState,
-}
-
-impl OutboundInfo {
-    async fn dnet_info(&self, p2p: P2pPtr) -> Option<Self> {
-        let addr = self.addr.clone()?;
-        let chan = p2p.channels().lock().await.get(&addr).cloned()?;
-        Some(Self { addr: Some(addr), channel: Some(chan.dnet_info().await), state: self.state })
-    }
-}
-
-impl Default for OutboundInfo {
-    fn default() -> Self {
-        Self { addr: None, channel: None, state: OutboundState::Open }
-    }
-}
-
 /// Defines outbound connections session.
 pub struct OutboundSession {
     /// Weak pointer to parent p2p object
@@ -108,8 +83,6 @@ pub struct OutboundSession {
     channel_subscriber: SubscriberPtr<Result<ChannelPtr>>,
     /// Flag to toggle channel_subscriber notifications
     notify: Mutex<bool>,
-    /// Channel debug info, corresponds to `connect_slots`
-    slot_info: Mutex<Vec<OutboundInfo>>,
 }
 
 impl OutboundSession {
@@ -120,7 +93,6 @@ impl OutboundSession {
             connect_slots: Mutex::new(vec![]),
             channel_subscriber: Subscriber::new(),
             notify: Mutex::new(false),
-            slot_info: Mutex::new(vec![]),
         })
     }
 
@@ -131,9 +103,6 @@ impl OutboundSession {
         // Activate mutex lock on connection slots.
         let mut connect_slots = self.connect_slots.lock().await;
 
-        // Create dnet stub
-        self.slot_info.lock().await.resize(n_slots, Default::default());
-
         for i in 0..n_slots {
             let task = StoppableTask::new();
 
@@ -249,11 +218,6 @@ impl OutboundSession {
                 // Remove pending lock since register_channel will add the channel to p2p
                 self.p2p().remove_pending(&addr).await;
 
-                dnet!(self,
-                    let info = &mut self.slot_info.lock().await[slot_number];
-                    info.state = OutboundState::Connected;
-                );
-
                 // Notify that channel processing has been finished
                 if *self.notify.lock().await {
                     self.channel_subscriber.notify(Ok(channel)).await;
@@ -276,12 +240,6 @@ impl OutboundSession {
         // At this point we failed to connect. We'll quarantine this peer now.
         self.p2p().hosts().quarantine(&addr).await;
 
-        dnet!(self,
-            let info = &mut self.slot_info.lock().await[slot_number];
-            info.addr = None;
-            info.state = OutboundState::Open;
-        );
-
         // Notify that channel processing failed
         if *self.notify.lock().await {
             self.channel_subscriber.notify(Err(Error::ConnectFailed)).await;
@@ -359,12 +317,6 @@ impl OutboundSession {
                     continue
                 }
 
-                dnet!(self,
-                    let info = &mut self.slot_info.lock().await[slot_number];
-                    info.addr = Some(host.clone());
-                    info.state = OutboundState::Pending;
-                );
-
                 return Ok(host.clone())
             }
 
@@ -456,12 +408,6 @@ impl OutboundSession {
     }
 }
 
-/// Dnet information for the outbound session
-pub struct OutboundDnet {
-    /// Slot information
-    pub slots: Vec<Option<OutboundInfo>>,
-}
-
 #[async_trait]
 impl Session for OutboundSession {
     fn p2p(&self) -> P2pPtr {
@@ -471,16 +417,4 @@ impl Session for OutboundSession {
     fn type_id(&self) -> SessionBitFlag {
         SESSION_OUTBOUND
     }
-
-    async fn dnet_info(&self) -> DnetInfo {
-        // We fetch channel infos for all outbound slots.
-        // If a slot is not connected, it will be `None`.
-        let mut slots = vec![];
-
-        for slot in self.slot_info.lock().await.iter() {
-            slots.push(slot.dnet_info(self.p2p()).await);
-        }
-
-        DnetInfo::Outbound(OutboundDnet { slots })
-    }
 }

src/net/session/seedsync_session.rs

@@ -47,7 +47,7 @@ use url::Url;
 use super::{
     super::{
         connector::Connector,
-        p2p::{DnetInfo, P2p, P2pPtr},
+        p2p::{P2p, P2pPtr},
     },
     Session, SessionBitFlag, SESSION_SEED,
 };
@@ -177,8 +177,4 @@ impl Session for SeedSyncSession {
     fn type_id(&self) -> SessionBitFlag {
         SESSION_SEED
     }
-
-    async fn dnet_info(&self) -> DnetInfo {
-        todo!()
-    }
 }

src/net/settings.rs

@@ -139,7 +139,7 @@ pub struct SettingsOpt {
     #[structopt(skip)]
     pub node_id: String,
 
-    /// Preferred transports for outbound connections    
+    /// Preferred transports for outbound connections
     #[serde(default)]
     #[structopt(long = "transports")]
     pub allowed_transports: Vec<String>,