sdk: require correlated COUNT results and preserve waiter receive errors

nostr-sdk/CHANGELOG.md

@@ -35,6 +35,11 @@
 - Add `LocalRelayBuilderNip42::relay_url` to make it possible to configure a custom `relay_url` for the nip42 challenge validation (https://github.com/nostrdevkit/nostr/pull/1476)
 - Add `LocalRelayBuilder::new_event_channel_size` for customizing the size of the channel used to notify new received events
 
+### Fixed
+
+- Require a correlated COUNT response before `Relay::count_events` returns a count.
+- Preserve the broadcast lag or closure cause in event `OK` and authentication waiters instead of returning a generic premature-exit error.
+
 ### Deprecated
 
 - Deprecate `LocalRelayBuilder::max_query_results` and `LocalRelayBuilder::default_filter_limit` (https://github.com/nostrdevkit/nostr/pull/1461)

nostr-sdk/src/relay/api/send_event.rs

@@ -114,7 +114,18 @@ impl EventSendStatus {
     }
 }
 
-/// Send event to relay
+/// Send an event to one relay.
+///
+/// By default, the operation waits for a matching `OK true` acknowledgement
+/// from the relay. A matching `OK false` is an explicit rejection.
+///
+/// Timeout, disconnection, notification loss, or notification closure while
+/// waiting for the acknowledgement leaves publication unconfirmed: the relay
+/// may have received the event. Callers should retain the event ID and resolve
+/// ambiguity according to their own policy.
+///
+/// Use [`SendEvent::wait_for_ok(false)`] to return after send submission without
+/// waiting for relay acknowledgement.
 #[must_use = "Does nothing unless you await!"]
 pub struct SendEvent<'relay, 'event> {
     relay: &'relay Relay,
@@ -135,7 +146,10 @@ impl<'relay, 'event> SendEvent<'relay, 'event> {
         }
     }
 
-    /// Wait for OK confirmation by the relay (default: true)
+    /// Wait for a matching `OK` confirmation by the relay (default: true).
+    ///
+    /// If disabled, [`EventSendStatus::Sent`] confirms only SDK send submission,
+    /// not relay acceptance.
     #[inline]
     pub fn wait_for_ok(mut self, enable: bool) -> Self {
         self.wait_for_ok = enable;
@@ -184,8 +198,8 @@ async fn wait_for_authentication(
     timeout: Duration,
 ) -> Result<(), Error> {
     time::timeout(Some(timeout), async {
-        while let Ok(notification) = notifications.recv().await {
-            match notification {
+        loop {
+            match notifications.recv().await.map_err(Error::from)? {
                 RelayNotification::Authenticated => {
                     return Ok(());
                 }
@@ -198,8 +212,6 @@ async fn wait_for_authentication(
                 _ => (),
             }
         }
-
-        Err(Error::state_msg("premature exit"))
     })
     .await
     .ok_or_else(Error::timeout)?
@@ -268,6 +280,7 @@ where
 
 #[cfg(test)]
 mod tests {
+    use std::error::Error as _;
     use std::time::Duration;
 
     use nostr::prelude::*;
@@ -392,4 +405,49 @@ mod tests {
         // Send as authenticated
         assert!(relay.send_event(&event).await.is_ok());
     }
+
+    #[tokio::test]
+    async fn authentication_waiter_preserves_receive_failure() {
+        let (tx, mut rx) = broadcast::channel(2);
+
+        for _ in 0..8 {
+            tx.send(RelayNotification::Authenticated).unwrap();
+        }
+
+        let error = wait_for_authentication(&mut rx, Duration::from_secs(1))
+            .await
+            .unwrap_err();
+
+        assert!(
+            error
+                .source()
+                .and_then(|source| source.downcast_ref::<broadcast::error::RecvError>())
+                .is_some_and(|source| matches!(source, broadcast::error::RecvError::Lagged(_)))
+        );
+
+        let (tx, mut rx) = broadcast::channel(2);
+
+        drop(tx);
+
+        let error = wait_for_authentication(&mut rx, Duration::from_secs(1))
+            .await
+            .unwrap_err();
+
+        assert!(
+            error
+                .source()
+                .and_then(|source| source.downcast_ref::<broadcast::error::RecvError>())
+                .is_some_and(|source| matches!(source, broadcast::error::RecvError::Closed))
+        );
+
+        let (tx, mut rx) = broadcast::channel(2);
+
+        tx.send(RelayNotification::AuthenticationFailed).unwrap();
+
+        let error = wait_for_authentication(&mut rx, Duration::from_secs(1))
+            .await
+            .unwrap_err();
+
+        assert_eq!(error.kind(), crate::error::ErrorKind::Rejected);
+    }
 }

nostr-sdk/src/relay/inner.rs

@@ -1470,8 +1470,8 @@ impl InnerRelay {
         timeout: Duration,
     ) -> Result<(bool, String), Error> {
         time::timeout(Some(timeout), async {
-            while let Ok(notification) = notifications.recv().await {
-                match notification {
+            loop {
+                match notifications.recv().await.map_err(Error::from)? {
                     RelayNotification::Message { message } => {
                         if let RelayMessage::Ok {
                             event_id,
@@ -1491,8 +1491,6 @@ impl InnerRelay {
                     _ => (),
                 }
             }
-
-            Err(Error::state_msg("premature exit"))
         })
         .await
         .ok_or_else(Error::timeout)?
@@ -1875,6 +1873,8 @@ async fn close_ws(tx: &mut WebSocketSink) -> Result<(), Error> {
 
 #[cfg(test)]
 mod tests {
+    use std::borrow::Cow;
+    use std::error::Error as _;
     use std::future::Future;
     use std::pin::Pin;
     use std::sync::Arc;
@@ -2274,6 +2274,90 @@ mod tests {
 
         assert!(!relay.inner.should_resubscribe(&subscription_id).await);
     }
+
+    #[tokio::test]
+    async fn ok_waiter_requires_matching_reply_and_preserves_receive_failure() {
+        let relay = Relay::new(RelayUrl::parse("wss://relay.example.com").unwrap());
+
+        let wanted = EventId::from_byte_array([0; 32]);
+        let other = EventId::from_byte_array([1; 32]);
+
+        let (tx, mut rx) = broadcast::channel(2);
+
+        tx.send(RelayNotification::Message {
+            message: Box::new(RelayMessage::Ok {
+                event_id: other,
+                status: true,
+                message: Cow::Borrowed("unrelated"),
+            }),
+        })
+        .unwrap();
+
+        tx.send(RelayNotification::Message {
+            message: Box::new(RelayMessage::Ok {
+                event_id: wanted,
+                status: false,
+                message: Cow::Borrowed("rejected"),
+            }),
+        })
+        .unwrap();
+
+        let (accepted, message) = relay
+            .inner
+            .wait_for_ok(&mut rx, &wanted, Duration::from_secs(1))
+            .await
+            .unwrap();
+
+        assert!(!accepted);
+        assert_eq!(message, "rejected");
+
+        let (_silent_tx, mut silent_rx) = broadcast::channel(2);
+
+        let error = relay
+            .inner
+            .wait_for_ok(&mut silent_rx, &wanted, Duration::from_millis(20))
+            .await
+            .unwrap_err();
+
+        assert_eq!(error.kind(), ErrorKind::Timeout);
+
+        let (lag_tx, _) = broadcast::channel(2);
+        let mut lagged = lag_tx.subscribe();
+
+        for _ in 0..8 {
+            lag_tx.send(RelayNotification::Authenticated).unwrap();
+        }
+
+        let error = relay
+            .inner
+            .wait_for_ok(&mut lagged, &wanted, Duration::from_secs(1))
+            .await
+            .unwrap_err();
+
+        assert!(
+            error
+                .source()
+                .and_then(|source| source.downcast_ref::<broadcast::error::RecvError>())
+                .is_some_and(|source| matches!(source, broadcast::error::RecvError::Lagged(_)))
+        );
+
+        let (closed_tx, mut closed_rx) = broadcast::channel(2);
+
+        drop(closed_tx);
+
+        let error = relay
+            .inner
+            .wait_for_ok(&mut closed_rx, &wanted, Duration::from_secs(1))
+            .await
+            .unwrap_err();
+
+        assert!(
+            error
+                .source()
+                .and_then(|source| source.downcast_ref::<broadcast::error::RecvError>())
+                .is_some_and(|source| matches!(source, broadcast::error::RecvError::Closed))
+        );
+    }
 }
 
 #[cfg(bench)]