Merge branch 'master' into cpp-bindings

Author: inetic

lib/src/network/connection.rs

@@ -1,7 +1,7 @@
 use super::{
     peer_addr::PeerAddr,
     peer_info::PeerInfo,
-    peer_source::PeerSource,
+    peer_source::{ConnectionDirection, PeerSource},
     peer_state::PeerState,
     runtime_id::PublicRuntimeId,
     stats::{ByteCounters, StatsTracker},
@@ -10,7 +10,6 @@ use crate::{
     collections::{hash_map::Entry, HashMap},
     sync::{AwaitDrop, DropAwaitable, WatchSenderExt},
 };
-use serde::Serialize;
 use std::{
     fmt,
     sync::{
@@ -40,7 +39,7 @@ impl ConnectionSet {
     pub fn reserve(&self, addr: PeerAddr, source: PeerSource) -> ReserveResult {
         let key = ConnectionKey {
             addr,
-            dir: ConnectionDirection::from_source(source),
+            dir: source.direction(),
         };
 
         self.connections
@@ -145,24 +144,6 @@ impl PeerInfoCollector {
     }
 }
 
-#[derive(Clone, Copy, Eq, PartialEq, Ord, PartialOrd, Hash, Debug, Serialize)]
-pub(super) enum ConnectionDirection {
-    Incoming,
-    Outgoing,
-}
-
-impl ConnectionDirection {
-    pub fn from_source(source: PeerSource) -> Self {
-        match source {
-            PeerSource::Listener => Self::Incoming,
-            PeerSource::UserProvided
-            | PeerSource::LocalDiscovery
-            | PeerSource::Dht
-            | PeerSource::PeerExchange => Self::Outgoing,
-        }
-    }
-}
-
 /// Connection permit that prevents another connection to the same peer (socket address) to be
 /// established as long as it remains in scope.
 pub(super) struct ConnectionPermit {

lib/src/network/connection_monitor.rs

@@ -1,8 +1,4 @@
-use super::{
-    connection::{ConnectionDirection, ConnectionId},
-    peer_addr::PeerAddr,
-    PeerSource, PublicRuntimeId,
-};
+use super::{connection::ConnectionId, peer_addr::PeerAddr, PeerSource, PublicRuntimeId};
 use crate::crypto::sign::PublicKey;
 use state_monitor::{MonitoredValue, StateMonitor};
 use std::sync::atomic::{AtomicU64, Ordering};
@@ -26,10 +22,7 @@ impl ConnectionMonitor {
             runtime_id = field::Empty,
         );
 
-        let direction_glyph = match ConnectionDirection::from_source(source) {
-            ConnectionDirection::Incoming => '↓',
-            ConnectionDirection::Outgoing => '↑',
-        };
+        let direction_glyph = source.direction().glyph();
 
         // We need to ID the StateMonitor node because it is created prior to `addr` being
         // deduplicated and so we'd get an ambiguous entry otherwise.

lib/src/network/message_broker.rs

@@ -10,12 +10,17 @@ use super::{
     server::Server,
     stats::ByteCounters,
 };
-use crate::{collections::HashMap, crypto::Hashable, protocol::RepositoryId, repository::Vault};
+use crate::{
+    collections::HashMap,
+    crypto::{sign::PublicKey, Hashable},
+    protocol::RepositoryId,
+    repository::Vault,
+};
 use backoff::{backoff::Backoff, ExponentialBackoffBuilder};
 use bytes::{BufMut, BytesMut};
 use futures_util::{SinkExt, StreamExt};
 use net::{bus::TopicId, unified::Connection};
-use state_monitor::StateMonitor;
+use state_monitor::{MonitoredValue, StateMonitor};
 use std::{collections::hash_map::Entry, sync::Arc, time::Instant};
 use tokio::{
     select,
@@ -36,6 +41,7 @@ pub(super) struct MessageBroker {
     links: HashMap<RepositoryId, oneshot::Sender<()>>,
     pex_peer: PexPeer,
     monitor: StateMonitor,
+    _monitor_runtime_id: MonitoredValue<PublicKey>,
     span: SpanGuard,
 }
 
@@ -51,6 +57,8 @@ impl MessageBroker {
     ) -> Self {
         let span = SpanGuard::new(Span::current());
 
+        let monitor_runtime_id = monitor.make_value("runtime id", *that_runtime_id.as_public_key());
+
         Self {
             this_runtime_id,
             that_runtime_id,
@@ -61,6 +69,7 @@ impl MessageBroker {
             links: HashMap::default(),
             pex_peer,
             monitor,
+            _monitor_runtime_id: monitor_runtime_id,
             span,
         }
     }

lib/src/network/mod.rs

@@ -41,29 +41,29 @@ pub use self::{
     runtime_id::{PublicRuntimeId, SecretRuntimeId},
     stats::Stats,
 };
-use choke::Choker;
-use constants::REQUEST_TIMEOUT;
-use event::ProtocolVersions;
 pub use net::stun::NatBehavior;
-use request_tracker::RequestTracker;
 
 use self::{
+    choke::Choker,
     connection::{ConnectionPermit, ConnectionSet, ReserveResult},
     connection_monitor::ConnectionMonitor,
+    constants::REQUEST_TIMEOUT,
     dht_discovery::DhtDiscovery,
+    event::ProtocolVersions,
     gateway::{Connectivity, Gateway, StackAddresses},
     local_discovery::LocalDiscovery,
     message_broker::MessageBroker,
     peer_addr::PeerPort,
     peer_exchange::{PexDiscovery, PexRepository},
+    peer_source::ConnectionDirection,
     protocol::{Version, MAGIC, VERSION},
+    request_tracker::RequestTracker,
     seen_peers::{SeenPeer, SeenPeers},
     stats::{ByteCounters, StatsTracker},
     stun::StunClients,
 };
 use crate::{
     collections::HashSet,
-    network::connection::ConnectionDirection,
     protocol::RepositoryId,
     repository::{RepositoryHandle, Vault},
 };
@@ -831,7 +831,7 @@ impl Inner {
             &connection,
             VERSION,
             &self.this_runtime_id,
-            ConnectionDirection::from_source(permit.source()),
+            permit.source().direction(),
         )
         .await;
 
@@ -883,8 +883,11 @@ impl Inner {
                     that_runtime_id,
                     connection,
                     pex_peer,
-                    self.peers_monitor
-                        .make_child(format!("{:?}", that_runtime_id.as_public_key())),
+                    self.peers_monitor.make_child(format!(
+                        "{} {}",
+                        permit.source().direction().glyph(),
+                        permit.addr()
+                    )),
                     self.stats_tracker.bytes.clone(),
                     permit.byte_counters(),
                 )

lib/src/network/peer_addr.rs

@@ -69,9 +69,7 @@ impl FromStr for PeerAddr {
         let (proto, addr) = match s.split_once('/') {
             Some((proto, addr)) => (proto, addr),
             None => {
-                return Err(format!(
-                    "Could not find '/' delimiter in the address {s:?}"
-                ));
+                return Err(format!("Could not find '/' delimiter in the address {s:?}"));
             }
         };
 

lib/src/network/peer_exchange.rs

@@ -2,10 +2,10 @@
 //! other in order to discover new peers.
 
 use super::{
-    connection::ConnectionDirection,
     ip,
     message::Message,
     peer_addr::PeerAddr,
+    peer_source::ConnectionDirection,
     seen_peers::{SeenPeer, SeenPeers},
     PeerSource,
 };
@@ -159,9 +159,7 @@ impl PexPeer {
     /// Call this whenever new connection to this peer has been established. The `closed` should
     /// trigger when the connection gets closed.
     pub fn handle_connection(&self, addr: PeerAddr, source: PeerSource, closed: AwaitDrop) {
-        if addr.is_tcp()
-            && ConnectionDirection::from_source(source) == ConnectionDirection::Incoming
-        {
+        if addr.is_tcp() && source.direction() == ConnectionDirection::Incoming {
             // Incomming TCP address can't be connected to (it has different port than the listener)
             // so there is no point exchanging it.
             return;

lib/src/network/peer_source.rs

@@ -33,6 +33,17 @@ pub enum PeerSource {
     PeerExchange = 4,
 }
 
+impl PeerSource {
+    pub(super) fn direction(&self) -> ConnectionDirection {
+        match self {
+            Self::Listener => ConnectionDirection::Incoming,
+            Self::UserProvided | Self::LocalDiscovery | Self::Dht | Self::PeerExchange => {
+                ConnectionDirection::Outgoing
+            }
+        }
+    }
+}
+
 impl fmt::Display for PeerSource {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         match self {
@@ -44,3 +55,20 @@ impl fmt::Display for PeerSource {
         }
     }
 }
+
+#[derive(Clone, Copy, Eq, PartialEq, Ord, PartialOrd, Hash, Debug, Serialize)]
+pub(super) enum ConnectionDirection {
+    /// Peer connected to us
+    Incoming,
+    /// We connected to the peer
+    Outgoing,
+}
+
+impl ConnectionDirection {
+    pub fn glyph(&self) -> char {
+        match self {
+            Self::Incoming => '↓',
+            Self::Outgoing => '↑',
+        }
+    }
+}

service/src/logger/mod.rs

@@ -49,7 +49,7 @@ fn create_log_filter() -> EnvFilter {
         // time (via a command-line option or so)
         .parse_lossy(
             env::var(EnvFilter::DEFAULT_ENV)
-                .unwrap_or_else(|_| "ouisync=debug,deadlock=warn".to_string()),
+                .unwrap_or_else(|_| "ouisync=debug,state_monitor=warn,deadlock=warn".to_string()),
         )
 }