sdk: report receiver-local notification gaps

nostr-sdk/CHANGELOG.md

@@ -31,6 +31,8 @@
 
 ### Added
 
+- Add `Client::notifications_with_gaps` and `Relay::notifications_with_gaps`
+  to report receiver-local notification loss while continuing reception.
 - Add `LocalRelay::connections_left` (https://github.com/nostrdevkit/nostr/pull/1459)
 - 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

nostr-sdk/Cargo.toml

@@ -75,6 +75,9 @@ name = "gossip"
 [[example]]
 name = "monitor"
 
+[[example]]
+name = "notification-gaps"
+
 [[example]]
 name = "nostr-connect"
 

nostr-sdk/README.md

@@ -15,6 +15,14 @@ The SDK can be used to build both sides of a nostr application:
 - clients, bots, and services that connect to existing relays;
 - local relays that run inside your process, including mock relays for tests.
 
+For notification consumers that need to detect receiver buffer overflow, use
+`Client::notifications_with_gaps` or `Relay::notifications_with_gaps`. A `Lagged`
+item reports the number of notifications skipped by that receiver; subsequent
+notifications remain available. The count is not a count of unique events, and
+the stream does not establish complete relay history or downstream storage.
+See the [notification gaps example](examples/notification-gaps.rs) for a minimal
+consumer that continues after a gap while marking its coverage invalid.
+
 ## Getting started
 
 ### Client

nostr-sdk/examples/notification-gaps.rs

@@ -0,0 +1,32 @@
+use nostr_sdk::prelude::*;
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let relay_url = std::env::args().nth(1).expect("pass a relay URL");
+    let client = Client::new();
+    client.add_relay(relay_url).and_connect().await?;
+
+    // Subscribe to notifications before asking for events.
+    let mut notifications = client.notifications_with_gaps();
+    client
+        .subscribe(Filter::new().kind(Kind::TextNote).since(Timestamp::now()))
+        .await?;
+
+    let mut coverage_invalid = false;
+    while let Some(update) = notifications.next().await {
+        match update {
+            NotificationUpdate::Lagged { skipped } => {
+                coverage_invalid = true;
+                eprintln!("This receiver missed {skipped} notifications; reacquisition is needed");
+            }
+            NotificationUpdate::Notification(ClientNotification::Event { event, .. }) => {
+                // Events after a gap are still useful, but they do not repair it.
+                println!("{} (coverage invalid: {coverage_invalid})", event.id);
+            }
+            NotificationUpdate::Notification(ClientNotification::Shutdown) => break,
+            NotificationUpdate::Notification(_) => {}
+        }
+    }
+
+    Ok(())
+}

nostr-sdk/src/client/mod.rs

@@ -34,7 +34,7 @@ use crate::proxy::Proxy;
 use crate::relay::{
     Relay, RelayCapabilities, RelayLimits, RelayOptions, SleepWhenIdle, SyncOptions,
 };
-use crate::stream::NotificationStream;
+use crate::stream::{NotificationStream, NotificationUpdate, ReportingNotificationStream};
 
 #[derive(Debug)]
 struct ClientConfig {
@@ -194,6 +194,9 @@ impl Client {
     ///
     /// The stream terminates when the client shutdowns.
     ///
+    /// This legacy stream silently skips notifications lost when its receiver
+    /// falls behind. Use [`Client::notifications_with_gaps`] when coverage matters.
+    ///
     /// <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 notifications(&self) -> Pin<Box<dyn Stream<Item = ClientNotification> + Send>> {
@@ -222,6 +225,42 @@ impl Client {
         )
     }
 
+    /// Get a notification stream that reports loss to this receiver.
+    ///
+    /// A [`NotificationUpdate::Lagged`] item means this receiver missed the
+    /// reported number of notifications. It does not count unique events or
+    /// identify affected subscriptions. Reception continues after a gap; callers
+    /// must explicitly reacquire any coverage they require.
+    ///
+    /// The stream starts at the moment this method is called. It emits
+    /// [`ClientNotification::Shutdown`] and then terminates on client shutdown.
+    /// A stream created after shutdown is empty. Dropping the stream ends only
+    /// this receiver; it does not stop the client.
+    #[inline]
+    pub fn notifications_with_gaps(
+        &self,
+    ) -> Pin<Box<dyn Stream<Item = NotificationUpdate<ClientNotification>> + Send>> {
+        if self.is_shutdown() {
+            return Box::pin(futures::stream::empty());
+        }
+
+        let rx = self.pool().notifications();
+        let (tx, rx_done) = oneshot::channel();
+        let mut tx: Option<oneshot::Sender<()>> = Some(tx);
+
+        Box::pin(
+            ReportingNotificationStream::new(rx)
+                .inspect(move |update| {
+                    if let NotificationUpdate::Notification(ClientNotification::Shutdown) = update {
+                        if let Some(tx) = tx.take() {
+                            let _ = tx.send(());
+                        }
+                    }
+                })
+                .take_until(rx_done),
+        )
+    }
+
     /// Get relays from the relay pool.
     ///
     /// # Configuration
@@ -1190,12 +1229,192 @@ impl Client {
 
 #[cfg(test)]
 mod tests {
+    use std::num::NonZeroUsize;
+
+    use nostr::event::{EventBuilder, FinalizeEvent, Kind};
+    use nostr::key::Keys;
     use nostr_gossip_memory::prelude::*;
 
     use super::*;
     use crate::error::ErrorKind;
     use crate::local_relay::*;
-    use crate::relay::RelayStatus;
+    use crate::relay::{RelayNotification, RelayStatus};
+
+    #[tokio::test]
+    async fn test_notification_gap_reporting_is_receiver_local() {
+        let local = LocalRelay::builder().build();
+        local.run().await.unwrap();
+
+        let url = local.url().await;
+        let client = Client::builder()
+            .notification_channel_size(NonZeroUsize::new(4).unwrap())
+            .build();
+        client
+            .add_relay(&url)
+            .notification_channel_size(4)
+            .and_connect()
+            .await
+            .unwrap();
+
+        let relay = client.relay(&url).await.unwrap().unwrap();
+        let mut fast_client = client.notifications_with_gaps();
+        let mut slow_client = client.notifications_with_gaps();
+        let mut fast_relay = relay.notifications_with_gaps();
+        let mut slow_relay = relay.notifications_with_gaps();
+        let keys = Keys::generate();
+        // Ignore another request's EOSE while waiting for the selected subscription.
+        client
+            .subscribe(Filter::new().kind(Kind::Metadata))
+            .await
+            .unwrap();
+        let id = client
+            .subscribe(Filter::new().author(keys.public_key()))
+            .await
+            .unwrap()
+            .value;
+
+        tokio::time::timeout(Duration::from_secs(5), async {
+            loop {
+                let update = fast_client.next().await.expect("active client stream closed");
+                assert!(matches!(update, NotificationUpdate::Notification(_)), "active client receiver lagged");
+                if matches!(update, NotificationUpdate::Notification(ClientNotification::Message { message, .. })
+                    if matches!(*message, RelayMessage::EndOfStoredEvents(ref received) if received.as_ref() == &id)) {
+                    break;
+                }
+            }
+        })
+        .await
+        .unwrap();
+        tokio::time::timeout(Duration::from_secs(5), async {
+            loop {
+                let update = fast_relay.next().await.expect("active relay stream closed");
+                assert!(matches!(update, NotificationUpdate::Notification(_)), "active relay receiver lagged");
+                if matches!(update, NotificationUpdate::Notification(RelayNotification::Message { message, .. })
+                    if matches!(*message, RelayMessage::EndOfStoredEvents(ref received) if received.as_ref() == &id)) {
+                    break;
+                }
+            }
+        })
+        .await
+        .unwrap();
+
+        let mut sent = Vec::new();
+        for n in 0..64 {
+            let event = EventBuilder::new(Kind::TextNote, format!("notification-{n}"))
+                .finalize(&keys)
+                .unwrap();
+            sent.push(event.id);
+            local.add_event(event).await.unwrap();
+
+            let received_client = tokio::time::timeout(Duration::from_secs(5), async {
+                loop {
+                    let update = fast_client
+                        .next()
+                        .await
+                        .expect("active client stream closed");
+                    assert!(
+                        matches!(update, NotificationUpdate::Notification(_)),
+                        "active client receiver lagged"
+                    );
+                    if let NotificationUpdate::Notification(ClientNotification::Event {
+                        event,
+                        ..
+                    }) = update
+                    {
+                        break event.id;
+                    }
+                }
+            })
+            .await
+            .unwrap();
+            let received_relay = tokio::time::timeout(Duration::from_secs(5), async {
+                loop {
+                    let update = fast_relay.next().await.expect("active relay stream closed");
+                    assert!(
+                        matches!(update, NotificationUpdate::Notification(_)),
+                        "active relay receiver lagged"
+                    );
+                    if let NotificationUpdate::Notification(RelayNotification::Event {
+                        event,
+                        ..
+                    }) = update
+                    {
+                        break event.id;
+                    }
+                }
+            })
+            .await
+            .unwrap();
+            assert_eq!(received_client, sent[n]);
+            assert_eq!(received_relay, sent[n]);
+        }
+
+        let first_client = tokio::time::timeout(Duration::from_secs(5), slow_client.next())
+            .await
+            .unwrap()
+            .unwrap();
+        let first_relay = tokio::time::timeout(Duration::from_secs(5), slow_relay.next())
+            .await
+            .unwrap()
+            .unwrap();
+        assert!(matches!(first_client, NotificationUpdate::Lagged { skipped } if skipped > 0));
+        assert!(matches!(first_relay, NotificationUpdate::Lagged { skipped } if skipped > 0));
+        assert!(matches!(
+            slow_client.next().await,
+            Some(NotificationUpdate::Notification(_))
+        ));
+        assert!(matches!(
+            slow_relay.next().await,
+            Some(NotificationUpdate::Notification(_))
+        ));
+
+        client.shutdown().await;
+
+        tokio::time::timeout(Duration::from_secs(5), async {
+            let mut saw_client_shutdown = false;
+            while let Some(update) = slow_client.next().await {
+                if matches!(
+                    update,
+                    NotificationUpdate::Notification(ClientNotification::Shutdown)
+                ) {
+                    saw_client_shutdown = true;
+                }
+            }
+            assert!(saw_client_shutdown);
+        })
+        .await
+        .unwrap();
+
+        tokio::time::timeout(Duration::from_secs(5), async {
+            let mut saw_relay_shutdown = false;
+            while let Some(update) = slow_relay.next().await {
+                if matches!(update, NotificationUpdate::Notification(RelayNotification::RelayStatus { status }) if status.is_shutdown()) {
+                    saw_relay_shutdown = true;
+                }
+            }
+            assert!(saw_relay_shutdown);
+        })
+        .await
+        .unwrap();
+    }
+
+    #[tokio::test]
+    async fn gap_streams_created_after_termination_are_empty() {
+        let client: Client = Client::new();
+        client.shutdown().await;
+        assert!(client.notifications_with_gaps().next().await.is_none());
+
+        for banned in [false, true] {
+            let relay =
+                crate::relay::Relay::new(RelayUrl::parse("wss://relay.example.com").unwrap());
+            if banned {
+                relay.ban();
+            } else {
+                relay.shutdown();
+            }
+            assert!(relay.notifications_with_gaps().next().await.is_none());
+        }
+    }
 
     #[tokio::test]
     async fn test_shutdown() {

nostr-sdk/src/lib.rs

@@ -24,3 +24,5 @@ mod stream;
 #[cfg(test)]
 mod test_utils;
 pub mod transport;
+
+pub use crate::stream::NotificationUpdate;

nostr-sdk/src/relay/mod.rs

@@ -40,7 +40,7 @@ pub use self::status::*;
 use crate::client::ClientNotification;
 use crate::error::Error;
 use crate::shared::SharedState;
-use crate::stream::NotificationStream;
+use crate::stream::{NotificationStream, NotificationUpdate, ReportingNotificationStream};
 
 /// Subscription auto-closed reason
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -213,6 +213,9 @@ impl Relay {
     ///
     /// The stream terminates when the relay shutdowns or is banned.
     ///
+    /// This legacy stream silently skips notifications lost when its receiver
+    /// falls behind. Use [`Relay::notifications_with_gaps`] when coverage matters.
+    ///
     /// <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 notifications(&self) -> Pin<Box<dyn Stream<Item = RelayNotification> + Send>> {
@@ -245,6 +248,43 @@ impl Relay {
         )
     }
 
+    /// Get a relay notification stream that reports loss to this receiver.
+    ///
+    /// A [`NotificationUpdate::Lagged`] item reports notifications skipped by
+    /// this receiver only. The stream continues after a gap. It starts at the
+    /// moment this method is called and terminates when the relay is shut down
+    /// or banned. Dropping it does not stop the relay.
+    #[inline]
+    pub fn notifications_with_gaps(
+        &self,
+    ) -> Pin<Box<dyn Stream<Item = NotificationUpdate<RelayNotification>> + Send>> {
+        let status = self.status();
+        if status.is_banned() || status.is_shutdown() {
+            return Box::pin(futures::stream::empty());
+        }
+
+        let rx = self.inner.internal_notification_sender.subscribe();
+        let (tx, rx_done) = oneshot::channel();
+        let mut tx: Option<oneshot::Sender<()>> = Some(tx);
+
+        Box::pin(
+            ReportingNotificationStream::new(rx)
+                .inspect(move |update| {
+                    if let NotificationUpdate::Notification(RelayNotification::RelayStatus {
+                        status,
+                    }) = update
+                    {
+                        if status.is_banned() || status.is_shutdown() {
+                            if let Some(tx) = tx.take() {
+                                let _ = tx.send(());
+                            }
+                        }
+                    }
+                })
+                .take_until(rx_done),
+        )
+    }
+
     /// Connect to the relay
     ///
     /// # Overview