lilith: Report inbound listener health

bin/lilith/src/main.rs

@@ -41,6 +41,7 @@ use darkfi::{
     async_daemonize, cli_desc,
     net::{
         self,
+        acceptor::InboundListenerHealth,
         hosts::HostColor,
         settings::{BanPolicy, MagicBytes, NetworkProfile},
         P2p, P2pPtr,
@@ -128,9 +129,19 @@ impl Spawn {
             addr_vec.push(JsonValue::String(addr.as_ref().to_string()));
         }
 
+        let listeners = self
+            .p2p
+            .session_inbound()
+            .listener_health()
+            .await
+            .iter()
+            .map(listener_health_info)
+            .collect();
+
         JsonValue::Object(HashMap::from([
             ("name".to_string(), JsonValue::String(self.name.clone())),
             ("urls".to_string(), JsonValue::Array(addr_vec)),
+            ("listeners".to_string(), JsonValue::Array(listeners)),
             ("whitelist".to_string(), JsonValue::Array(self.get_whitelist().await)),
             ("greylist".to_string(), JsonValue::Array(self.get_greylist().await)),
             ("goldlist".to_string(), JsonValue::Array(self.get_goldlist().await)),
@@ -138,6 +149,18 @@ impl Spawn {
     }
 }
 
+fn listener_health_info(health: &InboundListenerHealth) -> JsonValue {
+    JsonValue::Object(HashMap::from([
+        ("url".to_string(), JsonValue::String(health.url.to_string())),
+        ("running".to_string(), JsonValue::Boolean(health.running)),
+        ("active".to_string(), JsonValue::Number(health.active as f64)),
+        ("negotiating".to_string(), JsonValue::Number(health.negotiating as f64)),
+        ("limit".to_string(), JsonValue::Number(health.limit as f64)),
+        ("saturated".to_string(), JsonValue::Boolean(health.saturated())),
+        ("accept_backoff".to_string(), JsonValue::Boolean(health.accept_backoff)),
+    ]))
+}
+
 /// Defines the network-specific settings
 #[derive(Clone)]
 struct NetInfo {
@@ -509,9 +532,32 @@ async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
 
 #[cfg(test)]
 mod tests {
-    use darkfi::net::hosts::HostContainer;
+    use darkfi::net::{acceptor::InboundListenerHealth, hosts::HostContainer};
+    use tinyjson::JsonValue;
+    use url::Url;
 
-    use super::supported_network_profiles;
+    use super::{listener_health_info, supported_network_profiles};
+
+    #[test]
+    fn test_listener_health_info_reports_saturation() {
+        let health = InboundListenerHealth {
+            url: Url::parse("tor+tls://example.onion:9000").unwrap(),
+            running: true,
+            active: 2,
+            negotiating: 1,
+            limit: 3,
+            accept_backoff: false,
+        };
+
+        let info = listener_health_info(&health);
+        assert_eq!(info["url"], JsonValue::String(health.url.to_string()));
+        assert_eq!(info["running"], JsonValue::Boolean(true));
+        assert_eq!(info["active"], JsonValue::Number(2.0));
+        assert_eq!(info["negotiating"], JsonValue::Number(1.0));
+        assert_eq!(info["limit"], JsonValue::Number(3.0));
+        assert_eq!(info["saturated"], JsonValue::Boolean(true));
+        assert_eq!(info["accept_backoff"], JsonValue::Boolean(false));
+    }
 
     #[test]
     fn test_supported_network_profiles_are_consistent() {

src/net/acceptor.rs

@@ -19,7 +19,7 @@
 use std::{
     io::ErrorKind,
     sync::{
-        atomic::{AtomicUsize, Ordering::SeqCst},
+        atomic::{AtomicBool, AtomicUsize, Ordering::SeqCst},
         Arc,
     },
     time::Duration,
@@ -104,6 +104,35 @@ struct InboundSlotGuard {
     cv: Arc<CondVar>,
 }
 
+/// Current runtime state for an inbound listener.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct InboundListenerHealth {
+    pub url: Url,
+    pub running: bool,
+    pub active: usize,
+    pub negotiating: usize,
+    pub limit: usize,
+    pub accept_backoff: bool,
+}
+
+impl InboundListenerHealth {
+    pub fn saturated(&self) -> bool {
+        self.active + self.negotiating >= self.limit
+    }
+}
+
+/// Releases the transport-negotiation count when a handshake finishes or is cancelled.
+struct NegotiationGuard {
+    acceptor: AcceptorPtr,
+}
+
+impl Drop for NegotiationGuard {
+    fn drop(&mut self) {
+        let previous = self.acceptor.negotiating_count.fetch_sub(1, SeqCst);
+        debug_assert!(previous > 0, "inbound negotiation counter underflow");
+    }
+}
+
 impl InboundSlotGuard {
     fn new(acceptor: AcceptorPtr, cv: Arc<CondVar>) -> Self {
         Self { acceptor, cv }
@@ -120,29 +149,38 @@ impl Drop for InboundSlotGuard {
 
 /// Create inbound socket connections
 pub struct Acceptor {
+    endpoint: Url,
     channel_publisher: PublisherPtr<Result<ChannelPtr>>,
     task: StoppableTaskPtr,
     session: SessionWeakPtr,
     conn_count: AtomicUsize,
+    negotiating_count: AtomicUsize,
+    accept_backoff: AtomicBool,
+    running: AtomicBool,
     #[cfg(feature = "upnp-igd")]
     port_mappings: AsyncMutex<Vec<Arc<dyn PortMapping>>>,
 }
 
 impl Acceptor {
     /// Create new Acceptor object.
-    pub fn new(session: SessionWeakPtr) -> AcceptorPtr {
+    pub fn new(session: SessionWeakPtr, endpoint: Url) -> AcceptorPtr {
         Arc::new(Self {
+            endpoint,
             channel_publisher: Publisher::new(),
             task: StoppableTask::new(),
             session,
             conn_count: AtomicUsize::new(0),
+            negotiating_count: AtomicUsize::new(0),
+            accept_backoff: AtomicBool::new(false),
+            running: AtomicBool::new(false),
             #[cfg(feature = "upnp-igd")]
             port_mappings: AsyncMutex::new(Vec::new()),
         })
     }
 
     /// Start accepting inbound socket connections
-    pub async fn start(self: Arc<Self>, endpoint: Url, ex: ExecutorPtr) -> Result<()> {
+    pub async fn start(self: Arc<Self>, ex: ExecutorPtr) -> Result<()> {
+        let endpoint = self.endpoint.clone();
         let settings = self.session.upgrade().unwrap().p2p().settings();
         let settings = settings.read().await;
         let datastore = settings.p2p_datastore.clone();
@@ -203,6 +241,17 @@ impl Acceptor {
         self.channel_publisher.clone().subscribe().await
     }
 
+    pub fn health(&self, limit: usize) -> InboundListenerHealth {
+        InboundListenerHealth {
+            url: self.endpoint.clone(),
+            running: self.running.load(SeqCst),
+            active: self.conn_count.load(SeqCst),
+            negotiating: self.negotiating_count.load(SeqCst),
+            limit,
+            accept_backoff: self.accept_backoff.load(SeqCst),
+        }
+    }
+
     #[cfg(test)]
     pub(super) fn connection_count(&self) -> usize {
         self.conn_count.load(SeqCst)
@@ -215,6 +264,7 @@ impl Acceptor {
         handshake_timeout: Duration,
         ex: ExecutorPtr,
     ) {
+        self.running.store(true, SeqCst);
         let self_ = self.clone();
         self.task.clone().start(
             self.run_accept_loop(listener, handshake_timeout, ex.clone()),
@@ -224,6 +274,20 @@ impl Acceptor {
         );
     }
 
+    fn track_negotiation(
+        self: &Arc<Self>,
+        negotiation: PtNegotiation,
+        handshake_timeout: Duration,
+    ) -> PtNegotiation {
+        self.negotiating_count.fetch_add(1, SeqCst);
+        let guard = NegotiationGuard { acceptor: self.clone() };
+
+        Box::pin(async move {
+            let _guard = guard;
+            with_handshake_timeout(negotiation, handshake_timeout).await
+        })
+    }
+
     /// Run the accept loop.
     async fn run_accept_loop(
         self: Arc<Self>,
@@ -262,8 +326,9 @@ impl Acceptor {
                     match accept.await {
                         Ok(negotiation) => {
                             resource_backoff.reset();
+                            self.accept_backoff.store(false, SeqCst);
                             negotiations
-                                .push(with_handshake_timeout(negotiation, handshake_timeout));
+                                .push(self.track_negotiation(negotiation, handshake_timeout));
                             continue
                         }
                         Err(err) => Err(err),
@@ -275,8 +340,9 @@ impl Acceptor {
                     match select(accept, negotiation).await {
                         Either::Left((Ok(negotiation), _)) => {
                             resource_backoff.reset();
+                            self.accept_backoff.store(false, SeqCst);
                             negotiations
-                                .push(with_handshake_timeout(negotiation, handshake_timeout));
+                                .push(self.track_negotiation(negotiation, handshake_timeout));
                             continue
                         }
                         Either::Left((Err(err), _)) => Err(err),
@@ -293,6 +359,7 @@ impl Acceptor {
             } else if let Some(retry) = accept_retry.take() {
                 if negotiations.is_empty() {
                     retry.await;
+                    self.accept_backoff.store(false, SeqCst);
                     continue
                 }
 
@@ -300,7 +367,10 @@ impl Acceptor {
                 pin_mut!(negotiation);
 
                 match select(retry, negotiation).await {
-                    Either::Left((_, _)) => continue,
+                    Either::Left((_, _)) => {
+                        self.accept_backoff.store(false, SeqCst);
+                        continue
+                    }
                     Either::Right((Some(result), retry)) => {
                         accept_retry = Some(retry);
                         result
@@ -356,6 +426,7 @@ impl Acceptor {
 
                 Err(e) if is_descriptor_exhaustion(&e) => {
                     let delay = resource_backoff.next_delay();
+                    self.accept_backoff.store(true, SeqCst);
                     warn!(
                         target: "net::acceptor::run_accept_loop",
                         "[P2P] Listener descriptor exhaustion: {e}; retrying accepts in {} ms",
@@ -443,6 +514,8 @@ impl Acceptor {
     /// Handles network errors. Panics if errors pass silently, otherwise broadcasts it
     /// to all channel publishers.
     async fn handle_stop(self: Arc<Self>, result: Result<()>) {
+        self.running.store(false, SeqCst);
+        self.accept_backoff.store(false, SeqCst);
         match result {
             Ok(()) => panic!("Acceptor task should never complete without error status"),
             Err(err) => self.channel_publisher.notify(Err(err)).await,

src/net/session/inbound_session.rs

@@ -32,7 +32,7 @@ use url::Url;
 
 use super::{
     super::{
-        acceptor::{Acceptor, AcceptorPtr},
+        acceptor::{Acceptor, AcceptorPtr, InboundListenerHealth},
         channel::ChannelPtr,
         dnet::{self, dnetev, DnetEvent},
         p2p::{P2p, P2pPtr},
@@ -83,7 +83,7 @@ impl InboundSession {
         for (index, accept_addr) in inbound_addrs.iter().enumerate() {
             // First initialize an Acceptor and its Subscriber.
             let parent = Arc::downgrade(&self);
-            let acceptor = Acceptor::new(parent);
+            let acceptor = Acceptor::new(parent, accept_addr.clone());
 
             // Now start the Subscriber. The Subscriber will return a Channel once it has been
             // prepared by the Acceptor.
@@ -134,6 +134,12 @@ impl InboundSession {
         self.acceptors.lock().await.iter().map(|acceptor| acceptor.connection_count()).sum()
     }
 
+    /// Return a runtime health snapshot for every active inbound listener.
+    pub async fn listener_health(&self) -> Vec<InboundListenerHealth> {
+        let limit = self.p2p().settings().read().await.inbound_connections;
+        self.acceptors.lock().await.iter().map(|acceptor| acceptor.health(limit)).collect()
+    }
+
     /// Start accepting connections for inbound session.
     async fn start_accept_session(
         self: Arc<Self>,
@@ -144,7 +150,7 @@ impl InboundSession {
     ) -> Result<()> {
         info!(target: "net::inbound_session", "[P2P] Starting Inbound session #{index} on {accept_addr}");
         // Start listener
-        let result = acceptor.clone().start(accept_addr, ex).await;
+        let result = acceptor.clone().start(ex).await;
         if let Err(e) = &result {
             verbose!(target: "net::inbound_session", "[P2P] Error starting listener #{index}: {e}");
             acceptor.stop().await;

src/net/tests.rs

@@ -689,7 +689,23 @@ async fn p2p_tls_listener_accepts_while_handshake_stalled_real(ex: Arc<Executor<
 
     // Occupy the first accepted socket without sending a TLS ClientHello.
     let stalled = TcpStream::connect(&addr).await.unwrap();
-    Timer::after(Duration::from_millis(100)).await;
+    timeout(Duration::from_secs(1), async {
+        while p2p.session_inbound().listener_health().await[0].negotiating != 1 {
+            Timer::after(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .expect("listener health did not report the stalled TLS negotiation");
+
+    let health = p2p.session_inbound().listener_health().await;
+    assert_eq!(health.len(), 1);
+    assert_eq!(health[0].url, listen_url);
+    assert!(health[0].running);
+    assert_eq!(health[0].active, 0);
+    assert_eq!(health[0].negotiating, 1);
+    assert_eq!(health[0].limit, 2);
+    assert!(!health[0].saturated());
+    assert!(!health[0].accept_backoff);
 
     // A second client must complete TLS before the stalled handshake times out.
     let dialer = Dialer::new(listen_url, None, None, true).await.unwrap();