pool: add relay monitor

Introduce the `Monitor` struct to track relay status changes via channel notifications.

Closes #850

Pull-Request: #851
Signed-off-by: Yuki Kishimoto <[email protected]>

Author: yukibtc

CHANGELOG.md

@@ -44,6 +44,7 @@
 - mls-memory-storage: add an in-memory implementation for MLS ([JeffG] at https://github.com/rust-nostr/nostr/pull/839)
 - mls-sqlite-storage: a sqlite implementation for MLS ([JeffG] at https://github.com/rust-nostr/nostr/pull/842)
 - mls: add new crate for implementing MLS messaging ([JeffG] at https://github.com/rust-nostr/nostr/pull/843)
+- pool: add relay monitor ([Yuki Kishimoto] at https://github.com/rust-nostr/nostr/pull/851)
 - sdk: add `Options::pool` ([Yuki Kishimoto])
 
 ### Deprecated

crates/nostr-relay-pool/src/lib.rs

@@ -18,6 +18,7 @@ extern crate test;
 
 pub use async_wsocket::ConnectionMode;
 
+pub mod monitor;
 pub mod policy;
 pub mod pool;
 pub mod prelude;

crates/nostr-relay-pool/src/monitor.rs

@@ -0,0 +1,59 @@
+// Copyright (c) 2022-2023 Yuki Kishimoto
+// Copyright (c) 2023-2025 Rust Nostr Developers
+// Distributed under the MIT software license
+
+//! Monitor
+
+use nostr::RelayUrl;
+use tokio::sync::broadcast::{self, Receiver, Sender};
+
+use crate::relay::RelayStatus;
+
+/// Relay monitor notification
+#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub enum MonitorNotification {
+    /// Relay status changed
+    StatusChanged {
+        /// Relay URL
+        relay_url: RelayUrl,
+        /// Status
+        status: RelayStatus,
+    },
+}
+
+/// Relay monitor
+#[derive(Debug, Clone)]
+pub struct Monitor {
+    channel: Sender<MonitorNotification>,
+}
+
+impl Monitor {
+    /// Create a new monitor with the given channel size
+    ///
+    /// # Panics
+    ///
+    /// This will panic if the channel size is equal to `0` or larger than `usize::MAX / 2`;
+    pub fn new(channel_size: usize) -> Self {
+        let (tx, ..) = broadcast::channel(channel_size);
+
+        Self { channel: tx }
+    }
+
+    /// Subscribe to monitor notifications
+    ///
+    /// <div class="warning">When you call this method, you subscribe to the notifications channel from that precise moment. Anything received by relay/s before that moment is not included in the channel!</div>
+    #[inline]
+    pub fn subscribe(&self) -> Receiver<MonitorNotification> {
+        self.channel.subscribe()
+    }
+
+    #[inline]
+    fn notify(&self, notification: MonitorNotification) {
+        let _ = self.channel.send(notification);
+    }
+
+    #[inline]
+    pub(crate) fn notify_status_change(&self, relay_url: RelayUrl, status: RelayStatus) {
+        self.notify(MonitorNotification::StatusChanged { relay_url, status });
+    }
+}

crates/nostr-relay-pool/src/pool/builder.rs

@@ -11,6 +11,7 @@ use nostr_database::{MemoryDatabase, NostrDatabase};
 
 use super::options::RelayPoolOptions;
 use super::RelayPool;
+use crate::monitor::Monitor;
 use crate::policy::AdmitPolicy;
 use crate::transport::websocket::{DefaultWebsocketTransport, WebSocketTransport};
 
@@ -21,6 +22,8 @@ pub struct RelayPoolBuilder {
     pub websocket_transport: Arc<dyn WebSocketTransport>,
     /// Admission policy
     pub admit_policy: Option<Arc<dyn AdmitPolicy>>,
+    /// Relay monitor
+    pub monitor: Option<Monitor>,
     /// Relay pool options
     pub opts: RelayPoolOptions,
     // Private stuff
@@ -35,6 +38,7 @@ impl Default for RelayPoolBuilder {
         Self {
             websocket_transport: Arc::new(DefaultWebsocketTransport),
             admit_policy: None,
+            monitor: None,
             opts: RelayPoolOptions::default(),
             __database: Arc::new(MemoryDatabase::default()),
             __signer: None,
@@ -69,6 +73,13 @@ impl RelayPoolBuilder {
         self
     }
 
+    /// Set monitor
+    #[inline]
+    pub fn monitor(mut self, monitor: Monitor) -> Self {
+        self.monitor = Some(monitor);
+        self
+    }
+
     /// Set options
     #[inline]
     pub fn opts(mut self, opts: RelayPoolOptions) -> Self {

crates/nostr-relay-pool/src/pool/inner.rs

@@ -55,6 +55,7 @@ impl InnerRelayPool {
                 builder.__signer,
                 builder.admit_policy,
                 builder.opts.nip42_auto_authentication,
+                builder.monitor,
             ),
             atomic: Arc::new(AtomicPrivateData {
                 relays: RwLock::new(HashMap::new()),

crates/nostr-relay-pool/src/pool/mod.rs

@@ -28,6 +28,7 @@ pub use self::error::Error;
 use self::inner::{InnerRelayPool, Relays};
 pub use self::options::RelayPoolOptions;
 pub use self::output::Output;
+use crate::monitor::Monitor;
 use crate::relay::flags::FlagCheck;
 use crate::relay::options::{RelayOptions, ReqExitPolicy, SyncOptions};
 use crate::relay::Relay;
@@ -139,6 +140,13 @@ impl RelayPool {
         self.inner.notification_sender.subscribe()
     }
 
+    /// Returns the reference to the monitor, if any.
+    ///
+    /// Returns `None` if the monitor is not configured (see [`RelayPoolBuilder::monitor`] ).
+    pub fn monitor(&self) -> Option<&Monitor> {
+        self.inner.state.monitor.as_ref()
+    }
+
     /// Get shared state
     #[inline]
     pub fn state(&self) -> &SharedState {

crates/nostr-relay-pool/src/prelude.rs

@@ -14,6 +14,7 @@ pub use nostr::prelude::*;
 pub use nostr_database::*;
 
 // Internal modules
+pub use crate::monitor::{self, *};
 pub use crate::policy::*;
 pub use crate::pool::builder::*;
 pub use crate::pool::constants::*;

crates/nostr-relay-pool/src/relay/inner.rs

@@ -224,6 +224,11 @@ impl InnerRelay {
 
         // Send notification
         self.send_notification(RelayNotification::RelayStatus { status }, false);
+
+        // If monitor is enabled, notify status change.
+        if let Some(monitor) = &self.state.monitor {
+            monitor.notify_status_change(self.url.clone(), status);
+        }
     }
 
     /// Perform health checks

crates/nostr-relay-pool/src/shared.rs

@@ -15,6 +15,7 @@ use nostr::{EventId, NostrSigner};
 use nostr_database::{IntoNostrDatabase, MemoryDatabase, NostrDatabase};
 use tokio::sync::RwLock;
 
+use crate::monitor::Monitor;
 use crate::policy::AdmitPolicy;
 use crate::transport::websocket::{DefaultWebsocketTransport, WebSocketTransport};
 
@@ -47,6 +48,7 @@ pub struct SharedState {
     nip42_auto_authentication: Arc<AtomicBool>,
     verification_cache: Arc<Mutex<LruCache<u64, ()>>>,
     pub(crate) admit_policy: Option<Arc<dyn AdmitPolicy>>,
+    pub(crate) monitor: Option<Monitor>,
 }
 
 impl Default for SharedState {
@@ -57,6 +59,7 @@ impl Default for SharedState {
             None,
             None,
             true,
+            None,
         )
     }
 }
@@ -68,6 +71,7 @@ impl SharedState {
         signer: Option<Arc<dyn NostrSigner>>,
         admit_policy: Option<Arc<dyn AdmitPolicy>>,
         nip42_auto_authentication: bool,
+        monitor: Option<Monitor>,
     ) -> Self {
         let max_verification_cache_size: NonZeroUsize =
             NonZeroUsize::new(MAX_VERIFICATION_CACHE_SIZE)
@@ -80,6 +84,7 @@ impl SharedState {
             nip42_auto_authentication: Arc::new(AtomicBool::new(nip42_auto_authentication)),
             verification_cache: Arc::new(Mutex::new(LruCache::new(max_verification_cache_size))),
             admit_policy,
+            monitor,
         }
     }
 

crates/nostr-sdk/Cargo.toml

@@ -76,6 +76,9 @@ name = "fetch-events"
 name = "gossip"
 required-features = ["all-nips"]
 
+[[example]]
+name = "monitor"
+
 [[example]]
 name = "nostr-connect"
 required-features = ["nip59"]