relay-pool: add SendMessageOptions to Relay and RelayPool message sending APIs

Introduce `SendMessageOptions` and `SendMessagePolicy` to control message sending behaviors, enhancing flexibility and enabling policies like waiting for a send confirmation.

Signed-off-by: Yuki Kishimoto <[email protected]>

Author: yukibtc

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

@@ -30,7 +30,7 @@ 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::options::{RelayOptions, ReqExitPolicy, SendMessageOptions, SyncOptions};
 use crate::relay::Relay;
 use crate::shared::SharedState;
 use crate::stream::{BoxedStream, ReceiverStream};
@@ -653,13 +653,14 @@ impl RelayPool {
         &self,
         urls: I,
         msg: ClientMessage<'_>,
+        opts: SendMessageOptions,
     ) -> Result<Output<()>, Error>
     where
         I: IntoIterator<Item = U>,
         U: TryIntoUrl,
         Error: From<<U as TryIntoUrl>::Err>,
     {
-        self.batch_msg_to(urls, &[msg]).await
+        self.batch_msg_to(urls, &[msg], opts).await
     }
 
     /// Send multiple client messages at once to specific relays
@@ -669,6 +670,7 @@ impl RelayPool {
         &self,
         urls: I,
         msgs: &[ClientMessage<'_>],
+        opts: SendMessageOptions,
     ) -> Result<Output<()>, Error>
     where
         I: IntoIterator<Item = U>,
@@ -713,7 +715,7 @@ impl RelayPool {
         for url in set.into_iter() {
             let relay: &Relay = self.internal_relay(&relays, &url)?;
             urls.push(url);
-            futures.push(relay.batch_msg(msgs));
+            futures.push(relay.batch_msg(msgs, opts));
         }
 
         // Join futures

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

@@ -9,6 +9,7 @@ use core::time::Duration;
 
 pub(super) const WAIT_FOR_OK_TIMEOUT: Duration = Duration::from_secs(10);
 pub(super) const WAIT_FOR_AUTHENTICATION_TIMEOUT: Duration = Duration::from_secs(7);
+pub(super) const WAIT_FOR_MSG_SEND_CONFIRMATION_TIMEOUT: Duration = Duration::from_secs(10);
 pub(super) const DEFAULT_CONNECTION_TIMEOUT: Duration = Duration::from_secs(60);
 
 /// Relay default notification channel size

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

@@ -22,11 +22,14 @@ use tokio::sync::{broadcast, oneshot, Mutex, MutexGuard, Notify, RwLock, RwLockW
 use super::constants::{
     DEFAULT_CONNECTION_TIMEOUT, JITTER_RANGE, MAX_RETRY_INTERVAL, MIN_ATTEMPTS, MIN_SUCCESS_RATE,
     NEGENTROPY_BATCH_SIZE_DOWN, NEGENTROPY_FRAME_SIZE_LIMIT, NEGENTROPY_HIGH_WATER_UP,
-    NEGENTROPY_LOW_WATER_UP, PING_INTERVAL, SLEEP_INTERVAL, WAIT_FOR_OK_TIMEOUT,
-    WEBSOCKET_TX_TIMEOUT,
+    NEGENTROPY_LOW_WATER_UP, PING_INTERVAL, SLEEP_INTERVAL, WAIT_FOR_MSG_SEND_CONFIRMATION_TIMEOUT,
+    WAIT_FOR_OK_TIMEOUT, WEBSOCKET_TX_TIMEOUT,
 };
 use super::flags::AtomicRelayServiceFlags;
-use super::options::{RelayOptions, ReqExitPolicy, SubscribeAutoCloseOptions, SyncOptions};
+use super::options::{
+    RelayOptions, ReqExitPolicy, SendMessageOptions, SendMessagePolicy, SubscribeAutoCloseOptions,
+    SyncOptions,
+};
 use super::ping::PingTracker;
 use super::stats::RelayConnectionStats;
 use super::{
@@ -1324,11 +1327,19 @@ impl InnerRelay {
     }
 
     #[inline]
-    pub async fn send_msg(&self, msg: ClientMessage<'_>) -> Result<(), Error> {
-        self.batch_msg(&[msg]).await
+    pub async fn send_msg(
+        &self,
+        msg: ClientMessage<'_>,
+        opts: SendMessageOptions,
+    ) -> Result<(), Error> {
+        self.batch_msg(&[msg], opts).await
     }
 
-    pub async fn batch_msg(&self, msgs: &[ClientMessage<'_>]) -> Result<(), Error> {
+    pub async fn batch_msg(
+        &self,
+        msgs: &[ClientMessage<'_>],
+        opts: SendMessageOptions,
+    ) -> Result<(), Error> {
         // Check if relay is operational
         self.ensure_operational()?;
 
@@ -1347,30 +1358,45 @@ impl InnerRelay {
             return Err(Error::ReadDisabled);
         }
 
-        let (tx, rx) = oneshot::channel();
+        // Handle send policy
+        match opts.policy {
+            SendMessagePolicy::DontWait => {
+                // Send messages
+                self.atomic.channels.send_client_msgs(msgs, None)
+            }
+            SendMessagePolicy::WaitForSendConfirmation { timeout } => {
+                let (tx, rx) = oneshot::channel();
 
-        // Send messages
-        self.atomic.channels.send_client_msgs(msgs, Some(tx))?;
+                // Send messages
+                self.atomic.channels.send_client_msgs(msgs, Some(tx))?;
 
-        // Wait for confirmation and propagate error, if any
-        time::timeout(Some(Duration::from_secs(10)), rx)
-            .await
-            .ok_or(Error::Timeout)?
-            .map_err(|_| Error::CantReceiveSendConfirmation)
+                // Wait for confirmation and propagate error, if any
+                time::timeout(Some(timeout), rx)
+                    .await
+                    .ok_or(Error::Timeout)?
+                    .map_err(|_| Error::CantReceiveSendConfirmation)
+            }
+        }
     }
 
     async fn send_neg_msg(&self, id: &SubscriptionId, message: &str) -> Result<(), Error> {
-        self.send_msg(ClientMessage::NegMsg {
-            subscription_id: Cow::Borrowed(id),
-            message: Cow::Borrowed(message),
-        })
+        self.send_msg(
+            ClientMessage::NegMsg {
+                subscription_id: Cow::Borrowed(id),
+                message: Cow::Borrowed(message),
+            },
+            SendMessageOptions::default(),
+        )
         .await
     }
 
     async fn send_neg_close(&self, id: &SubscriptionId) -> Result<(), Error> {
-        self.send_msg(ClientMessage::NegClose {
-            subscription_id: Cow::Borrowed(id),
-        })
+        self.send_msg(
+            ClientMessage::NegClose {
+                subscription_id: Cow::Borrowed(id),
+            },
+            SendMessageOptions::default(),
+        )
         .await
     }
 
@@ -1387,8 +1413,11 @@ impl InnerRelay {
         let mut notifications = self.internal_notification_sender.subscribe();
 
         // Send the AUTH message
-        self.send_msg(ClientMessage::Auth(Cow::Borrowed(&event)))
-            .await?;
+        self.send_msg(
+            ClientMessage::Auth(Cow::Borrowed(&event)),
+            SendMessageOptions::default(),
+        )
+        .await?;
 
         // Wait for OK
         // The event ID is already checked in `wait_for_ok` method
@@ -1447,7 +1476,11 @@ impl InnerRelay {
         let subscriptions = self.subscriptions().await;
         for (id, filters) in subscriptions.into_iter() {
             if !filters.is_empty() && self.should_resubscribe(&id).await {
-                self.send_msg(ClientMessage::req(id, filters)).await?;
+                self.send_msg(
+                    ClientMessage::req(id, filters),
+                    SendMessageOptions::default(),
+                )
+                .await?;
             } else {
                 tracing::debug!("Skip re-subscription of '{id}'");
             }
@@ -1495,9 +1528,13 @@ impl InnerRelay {
             // Close subscription
             let send_result = if to_close {
                 tracing::debug!(id = %id, "Auto-closing subscription.");
-                relay
-                    .send_msg(ClientMessage::Close(Cow::Borrowed(&id)))
-                    .await
+                let msg: ClientMessage<'_> = ClientMessage::Close(Cow::Borrowed(&id));
+                let opts: SendMessageOptions = SendMessageOptions::default().policy(
+                    SendMessagePolicy::WaitForSendConfirmation {
+                        timeout: WAIT_FOR_MSG_SEND_CONFIRMATION_TIMEOUT,
+                    },
+                );
+                relay.send_msg(msg, opts).await
             } else {
                 Ok(())
             };
@@ -1653,7 +1690,7 @@ impl InnerRelay {
                                 subscription_id: Cow::Borrowed(id),
                                 filters: filters.iter().map(Cow::Borrowed).collect(),
                             };
-                            let _ = self.send_msg(msg).await;
+                            let _ = self.send_msg(msg, SendMessageOptions::default()).await;
                         }
                     }
                     RelayNotification::AuthenticationFailed => {
@@ -1743,7 +1780,8 @@ impl InnerRelay {
         }
 
         // Send CLOSE message
-        self.send_msg(ClientMessage::Close(id)).await
+        self.send_msg(ClientMessage::Close(id), SendMessageOptions::default())
+            .await
     }
 
     pub async fn unsubscribe(&self, id: &SubscriptionId) -> Result<(), Error> {
@@ -1843,7 +1881,8 @@ impl InnerRelay {
                 match self.state.database().event_by_id(&id).await {
                     Ok(Some(event)) => {
                         in_flight_up.insert(id);
-                        self.send_msg(ClientMessage::event(event)).await?;
+                        self.send_msg(ClientMessage::event(event), SendMessageOptions::default())
+                            .await?;
                         num_sent += 1;
                     }
                     Ok(None) => {
@@ -1923,8 +1962,14 @@ impl InnerRelay {
         self.add_auto_closing_subscription(down_sub_id.clone(), vec![filter.clone()])
             .await;
 
+        // Construct send msg options
+        let opts: SendMessageOptions =
+            SendMessageOptions::default().policy(SendMessagePolicy::WaitForSendConfirmation {
+                timeout: WAIT_FOR_MSG_SEND_CONFIRMATION_TIMEOUT,
+            });
+
         // Send msg
-        if let Err(e) = self.send_msg(msg).await {
+        if let Err(e) = self.send_msg(msg, opts).await {
             // Remove previously added subscription
             self.remove_subscription(down_sub_id).await;
 
@@ -1998,7 +2043,8 @@ impl InnerRelay {
             id_size: None,
             initial_message: Cow::Owned(hex::encode(initial_message)),
         };
-        self.send_msg(open_msg).await?;
+        self.send_msg(open_msg, SendMessageOptions::default())
+            .await?;
 
         // Check if negentropy is supported
         check_negentropy_support(&sub_id, opts, &mut temp_notifications).await?;
@@ -2085,8 +2131,11 @@ impl InnerRelay {
                                 self.remove_subscription(&down_sub_id).await;
 
                                 // Close subscription
-                                self.send_msg(ClientMessage::Close(Cow::Borrowed(&down_sub_id)))
-                                    .await?;
+                                self.send_msg(
+                                    ClientMessage::Close(Cow::Borrowed(&down_sub_id)),
+                                    SendMessageOptions::default(),
+                                )
+                                .await?;
                             }
                         }
                         RelayMessage::Closed {

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

@@ -35,8 +35,8 @@ pub use self::flags::{AtomicRelayServiceFlags, FlagCheck, RelayServiceFlags};
 use self::inner::InnerRelay;
 pub use self::limits::RelayLimits;
 pub use self::options::{
-    RelayOptions, ReqExitPolicy, SubscribeAutoCloseOptions, SubscribeOptions, SyncDirection,
-    SyncOptions, SyncProgress,
+    RelayOptions, ReqExitPolicy, SendMessageOptions, SendMessagePolicy, SubscribeAutoCloseOptions,
+    SubscribeOptions, SyncDirection, SyncOptions, SyncProgress,
 };
 pub use self::stats::RelayConnectionStats;
 pub use self::status::RelayStatus;
@@ -405,14 +405,22 @@ impl Relay {
 
     /// Send msg to relay
     #[inline]
-    pub async fn send_msg(&self, msg: ClientMessage<'_>) -> Result<(), Error> {
-        self.inner.send_msg(msg).await
+    pub async fn send_msg(
+        &self,
+        msg: ClientMessage<'_>,
+        opts: SendMessageOptions,
+    ) -> Result<(), Error> {
+        self.inner.send_msg(msg, opts).await
     }
 
     /// Send multiple [`ClientMessage`] at once
     #[inline]
-    pub async fn batch_msg(&self, msgs: &[ClientMessage<'_>]) -> Result<(), Error> {
-        self.inner.batch_msg(msgs).await
+    pub async fn batch_msg(
+        &self,
+        msgs: &[ClientMessage<'_>],
+        opts: SendMessageOptions,
+    ) -> Result<(), Error> {
+        self.inner.batch_msg(msgs, opts).await
     }
 
     async fn _send_event(
@@ -422,7 +430,10 @@ impl Relay {
     ) -> Result<(bool, String), Error> {
         // Send the EVENT message
         self.inner
-            .send_msg(ClientMessage::Event(Cow::Borrowed(event)))
+            .send_msg(
+                ClientMessage::Event(Cow::Borrowed(event)),
+                SendMessageOptions::default(),
+            )
             .await?;
 
         // Wait for OK
@@ -573,7 +584,11 @@ impl Relay {
             .await;
 
         // Send REQ message
-        if let Err(e) = self.inner.send_msg(msg).await {
+        if let Err(e) = self
+            .inner
+            .send_msg(msg, SendMessageOptions::default())
+            .await
+        {
             // Remove previously added subscription
             self.inner.remove_subscription(&id).await;
 
@@ -601,7 +616,9 @@ impl Relay {
         };
 
         // Send REQ message
-        self.inner.send_msg(msg).await?;
+        self.inner
+            .send_msg(msg, SendMessageOptions::default())
+            .await?;
 
         // No auto-close subscription: update subscription filter
         self.inner.update_subscription(id, filters, true).await;
@@ -689,7 +706,9 @@ impl Relay {
             subscription_id: Cow::Borrowed(&id),
             filter: Cow::Owned(filter),
         };
-        self.inner.send_msg(msg).await?;
+        self.inner
+            .send_msg(msg, SendMessageOptions::default())
+            .await?;
 
         let mut count = 0;
 
@@ -715,7 +734,9 @@ impl Relay {
         .ok_or(Error::Timeout)?;
 
         // Unsubscribe
-        self.inner.send_msg(ClientMessage::close(id)).await?;
+        self.inner
+            .send_msg(ClientMessage::close(id), SendMessageOptions::default())
+            .await?;
 
         Ok(count)
     }

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

@@ -166,6 +166,34 @@ impl RelayOptions {
     }
 }
 
+/// Send message policy
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub enum SendMessagePolicy {
+    /// Don't wait for the send confirmation.
+    #[default]
+    DontWait,
+    /// Wait for send confirmation.
+    WaitForSendConfirmation {
+        /// Timeout
+        timeout: Duration,
+    },
+}
+
+/// Send message options
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub struct SendMessageOptions {
+    /// Policy to use when sending messages.
+    pub policy: SendMessagePolicy,
+}
+
+impl SendMessageOptions {
+    /// Set policy
+    pub const fn policy(mut self, policy: SendMessagePolicy) -> Self {
+        self.policy = policy;
+        self
+    }
+}
+
 /// Auto-closing subscribe options
 #[derive(Debug, Clone, Copy, Default)]
 pub struct SubscribeAutoCloseOptions {